Implement save transfer

Add ability to move or copy saves between instances.  For instances on
the same slave the saves are transferred with filesystem operations, for
instances on two different slaves the proxy stream interface is used to
send the save via the master server.

CHANGELOG.md

@@ -19,7 +19,7 @@
 - Packaged and uploaded to npm again, greatly simplifying distribution and installation of both Clusterio and plugins.
 - Added centralized logging for the cluster.
   All logs from both Clusterio and Factorio is stored in a shared log which can be inspected and queried.
-- Added remote save management with the ability to list, create, rename, copy, upload, download, delete and load saves for instances.
+- Added remote save management with the ability to list, create, rename, copy, upload, download, transfer, delete and load saves for instances.
 
 ### Features
 

packages/ctl/ctl.js

@@ -659,6 +659,34 @@ instanceSaveCommands.add(new libCommand.Command({
 	},
 }));
 
+instanceSaveCommands.add(new libCommand.Command({
+	definition: [
+		"transfer <source-instance> <source-save> <target-instance> [target-save]",
+		"Transfer a save between instances",
+		(yargs) => {
+			yargs.positional("source-instance", { describe: "Instance to transfer save from", type: "string" });
+			yargs.positional("source-save", { describe: "Save to transfer.", type: "string" });
+			yargs.positional("target-instance", { describe: "Instance to transfer to", type: "string" });
+			yargs.positional("target-save", { describe: "Name to give transferred save.", type: "string" });
+			yargs.options({
+				"copy": { describe: "Copy instead of moving the save", type: "boolean", default: false },
+			});
+		},
+	],
+	handler: async function(args, control) {
+		let sourceInstanceId = await libCommand.resolveInstance(control, args.sourceInstance);
+		let targetInstanceId = await libCommand.resolveInstance(control, args.targetInstance);
+		let result = await libLink.messages.transferSave.send(control, {
+			instance_id: sourceInstanceId,
+			source_save: args.sourceSave,
+			target_instance_id: targetInstanceId,
+			target_save: args.targetSave || args.sourceSave,
+			copy: args.copy,
+		});
+		print(`Transferred as ${result.save} to ${args.targetInstance}.`);
+	},
+}));
+
 instanceSaveCommands.add(new libCommand.Command({
 	definition: ["download <instance> <save>", "Download a save from an instance", (yargs) => {
 		yargs.positional("instance", { describe: "Instance to download save from", type: "string" });

packages/lib/link/messages.js

@@ -595,6 +595,22 @@ messages.downloadSave = new Request({
 	},
 });
 
+messages.transferSave = new Request({
+	type: "transfer_save",
+	links: ["control-master", "master-slave"],
+	permission: "core.instance.save.transfer",
+	requestProperties: {
+		"instance_id": { type: "integer" },
+		"source_save": { type: "string" },
+		"target_instance_id": { type: "integer" },
+		"target_save": { type: "string" },
+		"copy": { type: "boolean" },
+	},
+	responseProperties: {
+		"save": { type: "string" },
+	},
+});
+
 messages.pullSave = new Request({
 	type: "pull_save",
 	links: ["master-slave"],

packages/lib/users.js

@@ -412,6 +412,11 @@ definePermission({
 	title: "Upload save",
 	description: "Upload savegames to instances.",
 });
+definePermission({
+	name: "core.instance.save.transfer",
+	title: "Transfer save",
+	description: "Transfer savegames between instances.",
+});
 definePermission({
 	name: "core.instance.save.download",
 	title: "Download save",

packages/master/src/ControlConnection.js

@@ -1,5 +1,6 @@
 "use strict";
 const jwt = require("jsonwebtoken");
+const events = require("events");
 
 const libConfig = require("@clusterio/lib/config");
 const libErrors = require("@clusterio/lib/errors");
@@ -373,6 +374,85 @@ class ControlConnection extends BaseConnection {
 		return { stream_id: stream.id };
 	}
 
+	async transferSaveRequestHandler(message, request) {
+		let { source_save, target_save, copy, instance_id, target_instance_id } = message.data;
+		if (copy) {
+			this.user.checkPermission("core.instance.save.copy");
+		} else if (source_save !== target_save) {
+			this.user.checkPermission("core.instance.save.rename");
+		}
+
+		if (instance_id === target_instance_id) {
+			throw new libErrors.RequestError("Source and target instance may not be the same");
+		}
+		let sourceInstance = this._master.getRequestInstance(instance_id);
+		let targetInstance = this._master.getRequestInstance(target_instance_id);
+		let sourceSlaveId = sourceInstance.config.get("instance.assigned_slave");
+		let targetSlaveId = targetInstance.config.get("instance.assigned_slave");
+		if (sourceSlaveId === null) {
+			throw new libErrors.RequestError("Source instance is not assigned a slave");
+		}
+		if (targetSlaveId === null) {
+			throw new libErrors.RequestError("Target instance is not assigned a slave");
+		}
+
+		// Let slave handle request if source and target is on the same slave.
+		if (sourceSlaveId === targetSlaveId) {
+			return await this.forwardRequestToInstance(message, request);
+		}
+
+		// Check connectivity
+		let sourceSlaveConnection = this._master.wsServer.slaveConnections.get(sourceSlaveId);
+		if (!sourceSlaveConnection || sourceSlaveConnection.closing) {
+			throw new libErrors.RequestError("Source slave is not connected to the master server");
+		}
+
+		let targetSlaveConnection = this._master.wsServer.slaveConnections.get(targetSlaveId);
+		if (!targetSlaveConnection || targetSlaveConnection.closing) {
+			throw new libErrors.RequestError("Target slave is not connected to the master server");
+		}
+
+		// Create stream to proxy from target to source
+		let stream = await routes.createProxyStream(this._master.app);
+		stream.events.on("timeout", () => {
+			if (stream.source) {
+				stream.source.destroy();
+			}
+			stream.events.emit("error", new libErrors.RequestError("Timed out establishing transfer stream"));
+		});
+
+		// Ignore errors if not listening for them to avoid crash.
+		stream.events.on("error", () => { /* ignore */ });
+
+		// Establish push from source slave to stream, this is done first to
+		// ensure the file size is known prior to the target slave pull.
+		await Promise.all([
+			this._master.forwardRequestToInstance(libLink.messages.pushSave, {
+				instance_id,
+				stream_id: stream.id,
+				save: source_save,
+			}),
+			events.once(stream.events, "source"),
+		]);
+
+		// Establish pull from target slave to stream and wait for completion.
+		let result = await this._master.forwardRequestToInstance(libLink.messages.pullSave, {
+			instance_id: target_instance_id,
+			stream_id: stream.id,
+			filename: target_save,
+		});
+
+		// Delete source save if this is not a copy
+		if (!copy) {
+			await this._master.forwardRequestToInstance(libLink.messages.deleteSave, {
+				instance_id,
+				save: source_save,
+			});
+		}
+
+		return { save: result.save };
+	}
+
 	async setLogSubscriptionsRequestHandler(message) {
 		this.logSubscriptions = message.data;
 		this.updateLogSubscriptions();

packages/slave/slave.js

@@ -1489,6 +1489,45 @@ class Slave extends libLink.Link {
 		await this.sendSaveListUpdate(instance_id, path.join(instanceInfo.path, "saves"));
 	}
 
+	async transferSaveRequestHandler(message) {
+		let { source_save, target_save, copy, instance_id, target_instance_id } = message.data;
+		checkRequestSaveName(source_save);
+		checkRequestSaveName(target_save);
+		let sourceInstanceInfo = this.getRequestInstanceInfo(instance_id);
+		let targetInstanceInfo = this.getRequestInstanceInfo(target_instance_id);
+
+		// For consistency with remote transfer initiated through pullSave the
+		// target is renamed if it already exists.
+		target_save = await libFileOps.findUnusedName(
+			path.join(targetInstanceInfo.path, "saves"), target_save, ".zip"
+		);
+
+		try {
+			if (copy) {
+				await fs.copy(
+					path.join(sourceInstanceInfo.path, "saves", source_save),
+					path.join(targetInstanceInfo.path, "saves", target_save),
+					{ overwrite: true },
+				);
+			} else {
+				await fs.move(
+					path.join(sourceInstanceInfo.path, "saves", source_save),
+					path.join(targetInstanceInfo.path, "saves", target_save),
+					{ overwrite: true },
+				);
+			}
+		} catch (err) {
+			if (err.code === "ENOENT") {
+				throw new libErrors.RequestError(`${source_save} does not exist`);
+			}
+			throw err;
+		}
+		await this.sendSaveListUpdate(instance_id, path.join(sourceInstanceInfo.path, "saves"));
+		await this.sendSaveListUpdate(target_instance_id, path.join(targetInstanceInfo.path, "saves"));
+
+		return { save: target_save };
+	}
+
 	async sendSaveListUpdate(instance_id, savesDir) {
 		let instanceConnection = this.instanceConnections.get(instance_id);
 		let saveList;

packages/web_ui/src/components/SavesList.jsx

@@ -1,5 +1,8 @@
 import React, { useContext, useState } from "react";
-import { Button, Form, Input, List, Modal, Popconfirm, Progress, Space, Table, Tooltip, Upload } from "antd";
+import {
+	message, Button, Checkbox, Form, Input, List, Modal,
+	Popconfirm, Progress, Select, Space, Table, Tooltip, Upload,
+} from "antd";
 import CaretLeftOutlined from "@ant-design/icons/CaretLeftOutlined";
 import LeftOutlined from "@ant-design/icons/LeftOutlined";
 
@@ -9,6 +12,7 @@ import { useAccount } from "../model/account";
 import ControlContext from "./ControlContext";
 import CreateSaveModal from "./CreateSaveModal";
 import SectionHeader from "./SectionHeader";
+import { useInstanceList } from "../model/instance";
 import { useSaves } from "../model/saves";
 import { notifyErrorHandler } from "../util/notify";
 
@@ -91,6 +95,90 @@ function CopyModal(props) {
 	</>;
 }
 
+function TransferModal(props) {
+	let account = useAccount();
+	let control = useContext(ControlContext);
+	let [visible, setVisible] = useState(false);
+	let [form] = Form.useForm();
+	let [instanceList] = useInstanceList();
+
+	return <>
+		<Button disabled={props.disabled} onClick={() => setVisible(true)}>Transfer</Button>
+		<Modal
+			title="Transfer save"
+			okText="Transfer"
+			visible={visible}
+			onOk={() => form.submit()}
+			onCancel={() => setVisible(false)}
+			destroyOnClose
+		>
+			<Form
+				form={form}
+				labelCol={{ span: 6 }}
+				wrapperCol={{ span: 18 }}
+				initialValues={{ transferredName: props.save.name, copy: false }}
+				onFinish={values => {
+					setVisible(false);
+					let hide = message.loading("Transferring save...", 0);
+					libLink.messages.transferSave.send(control, {
+						instance_id: props.instanceId,
+						source_save: props.save.name,
+						target_instance_id: values.targetInstanceId,
+						target_save: values.transferredName || props.save.name,
+						copy: values.copy,
+					}).then(() => {
+						message.success("Transfer complete");
+						form.resetFields();
+					}).catch(
+						notifyErrorHandler("Error transfering save")
+					).finally(() => hide());
+				}}
+			>
+				<Form.Item
+					name="targetInstanceId"
+					label="Target Instance"
+					rules={[{ required: true, message: "Target Instance is required" }]}
+				>
+					<Select
+						autoFocus
+						showSearch
+						filterOption={(input, option) => option.title.toLowerCase().indexOf(input.toLowerCase()) >= 0}
+					>
+						{instanceList.filter(
+							instance => instance["id"] !== props.instanceId
+						).map((instance) => <Select.Option
+							key={instance["id"]}
+							value={instance["id"]}
+							title={instance["name"]}
+							disabled={["unassigned", "unknown"].includes(instance["status"])}
+						>
+							{instance["name"]}
+							{instance["status"] === "unassigned" && " (unassigned)"}
+							{instance["status"] === "unknown" && " (offline)"}
+						</Select.Option>)}
+					</Select>
+				</Form.Item>
+				<Form.Item
+					name="transferredName"
+					label="Transferred name"
+					disabled={!account.hasAnyPermission("core.instance.save.rename", "core.instance.save.copy")}
+				>
+					<Input />
+				</Form.Item>
+				<Form.Item
+					name="copy"
+					valuePropName="checked"
+					label="Copy"
+					tooltip="Copy instead of moving the save to the new instance."
+					disabled={!account.hasPermission("core.instance.save.copy")}
+				>
+					<Checkbox />
+				</Form.Item>
+			</Form>
+		</Modal>
+	</>;
+}
+
 function formatBytes(bytes) {
 	if (bytes === 0) {
 		return "0 Bytes";
@@ -179,6 +267,9 @@ export default function SavesList(props) {
 						);
 					}}
 				>Download</Button>}
+				{account.hasPermission("core.instance.save.transfer") && <TransferModal
+					disabled={slaveOffline} instanceId={props.instance.id} save={save}
+				/>}
 				{account.hasPermission("core.instance.save.delete") && <Popconfirm
 					title="Permanently delete save?"
 					okText="Delete"

test/integration/clusterio.js

@@ -3,6 +3,7 @@ const assert = require("assert").strict;
 const fs = require("fs-extra");
 const path = require("path");
 const events = require("events");
+const phin = require("phin");
 
 const libLink = require("@clusterio/lib/link");
 const libUsers = require("@clusterio/lib/users");
@@ -25,6 +26,24 @@ async function checkInstanceStatus(id, status) {
 	assert.equal(instances.get(44).status, status, "incorrect instance status");
 }
 
+async function uploadSave(instanceId, name, content) {
+	return await phin({
+		url: `https://localhost:4443/api/upload-save?instance_id=${instanceId}&filename=${name}`,
+		method: "POST",
+		core: { rejectUnauthorized: false },
+		headers: {
+			"X-Access-Token": controlToken,
+			"Content-Type": "application/zip",
+			"Content-Length": String(content.length),
+		},
+		data: content,
+	});
+}
+
+async function deleteSave(instanceId, save) {
+	await libLink.messages.deleteSave.send(getControl(), { instance_id: instanceId, save });
+}
+
 describe("Integration of Clusterio", function() {
 	describe("clusteriomaster", function() {
 		describe("bootstrap generate-user-token", function() {
@@ -115,10 +134,14 @@ describe("Integration of Clusterio", function() {
 				let config = "alt-slave-config.json";
 				let configPath = path.join("temp", "test", config);
 				await fs.remove(configPath);
+				await fs.remove(path.join("temp", "test", "alt-instances"));
 				await execCtl(`slave create-config --id 5 --name alt-slave --generate-token --output ${config}`);
 				await exec(
 					`node ../../packages/slave --config ${config} config set slave.tls_ca ../../test/file/tls/cert.pem`
 				);
+				await exec(
+					`node ../../packages/slave --config ${config} config set slave.instances_directory alt-instances`
+				);
 
 				let slaveProcess;
 				try {
@@ -598,6 +621,127 @@ describe("Integration of Clusterio", function() {
 			});
 		});
 
+		describe("instance save transfer", function() {
+			before(async function() {
+				slowTest(this);
+				await execCtl("instance create spam --id 66");
+				await execCtl("instance assign spam 4");
+				await execCtl("instance create unassign --id 77");
+			});
+			after(async function() {
+				await execCtl("instance delete spam");
+			});
+			for (let remote of [false, true]) {
+				let pri = 44;
+				let sec = remote ? 99 : 66;
+				let priSaves = path.join("temp", "test", "instances", "test", "saves");
+				let secSaves = remote
+					? path.join("temp", "test", "alt-instances", "alt-test", "saves")
+					: path.join("temp", "test", "instances", "spam", "saves")
+				;
+				describe(remote ? "remote" : "local", function() {
+					if (remote) {
+						let slaveProcess;
+						before(async function() {
+							// On windows there's currently no way to automate graceful shutdown of the slave
+							// process as CTRL+C is some weird terminal thing and SIGINT isn't a thing.
+							if (process.platform === "win32") {
+								this.skip();
+							}
+							slowTest(this);
+							// Reuse from the clusterioslave test
+							let config = "alt-slave-config.json";
+							slaveProcess = await spawn(
+								"alt-slave:", `node ../../packages/slave run --config ${config}`, /Started slave/
+							);
+						});
+						after(async function() {
+							if (slaveProcess) {
+								slaveProcess.kill("SIGINT");
+								await events.once(slaveProcess, "exit");
+							}
+						});
+					}
+					it("should transfers a save", async function() {
+						await uploadSave(pri, "transfer.zip", "transfer.zip");
+						await execCtl(`instance save transfer ${pri} transfer.zip ${sec}`);
+						assert(!await fs.pathExists(path.join(priSaves, "transfer.zip")), "save still at pri");
+						assert(await fs.pathExists(path.join(secSaves, "transfer.zip")), "save not at sec");
+						await deleteSave(sec, "transfer.zip");
+					});
+					it("should support rename", async function() {
+						await uploadSave(pri, "transfer.zip", "transfer.zip");
+						await execCtl(`instance save transfer ${pri} transfer.zip ${sec} rename.zip`);
+						assert(!await fs.pathExists(path.join(priSaves, "transfer.zip")), "save still at pri");
+						assert(await fs.pathExists(path.join(secSaves, "rename.zip")), "save not at sec");
+						await deleteSave(sec, "rename.zip");
+					});
+					it("should auto-rename if target exists", async function() {
+						await uploadSave(pri, "transfer.zip", "transfer.zip");
+						await uploadSave(sec, "transfer.zip", "transfer.zip");
+						await execCtl(`instance save transfer ${pri} transfer.zip ${sec}`);
+						assert(!await fs.pathExists(path.join(priSaves, "transfer.zip")), "save still at pri");
+						assert(await fs.pathExists(path.join(secSaves, "transfer-2.zip")), "save not at sec");
+						await deleteSave(sec, "transfer.zip");
+						await deleteSave(sec, "transfer-2.zip");
+					});
+					it("should copy when using --copy", async function() {
+						await uploadSave(pri, "transfer.zip", "transfer.zip");
+						await execCtl(`instance save transfer ${pri} transfer.zip ${sec} --copy`);
+						assert(await fs.pathExists(path.join(priSaves, "transfer.zip")), "save not at pri");
+						assert(await fs.pathExists(path.join(secSaves, "transfer.zip")), "save not at sec");
+						await deleteSave(pri, "transfer.zip");
+						await deleteSave(sec, "transfer.zip");
+					});
+					it("should fail if save does not exist", async function() {
+						await assert.rejects(execCtl(`instance save transfer ${pri} not-here.zip ${sec}`));
+					});
+					it("should fail if source save name is invalid", async function() {
+						await assert.rejects(execCtl(`instance save transfer ${pri} nul 111`));
+					});
+					it("should fail if target save name is invalid", async function() {
+						await uploadSave(pri, "transfer.zip", "transfer.zip");
+						await assert.rejects(execCtl(`instance save transfer ${pri} transfer.zip 111 nul`));
+						await deleteSave(pri, "transfer.zip");
+					});
+					it("should reject path traversal of source save", async function() {
+						await assert.rejects(execCtl(`instance save transfer ${pri} ../saves/transfer.zip 111`));
+					});
+					it("should reject path traversal of target save", async function() {
+						await uploadSave(pri, "transfer.zip", "transfer.zip");
+						await assert.rejects(
+							execCtl(`instance save transfer ${pri} transfer.zip 111 ../saves/transfer.zip`)
+						);
+						await deleteSave(pri, "transfer.zip");
+					});
+					if (!remote) {
+						it("should fail if source and target instance is the same instance", async function() {
+							await uploadSave(pri, "transfer.zip", "transfer.zip");
+							await assert.rejects(execCtl(`instance save transfer ${pri} transfer.zip ${pri}`));
+							await deleteSave(pri, "transfer.zip");
+						});
+						it("should fail if source instance does not exist", async function() {
+							await assert.rejects(execCtl(`instance save transfer 111 not-here.zip ${sec}`));
+						});
+						it("should fail if target instance does not exist", async function() {
+							await uploadSave(pri, "transfer.zip", "transfer.zip");
+							await assert.rejects(execCtl(`instance save transfer ${pri} transfer.zip 111`));
+							await deleteSave(pri, "transfer.zip");
+						});
+						it("should fail if source instance is not assigned", async function() {
+							await assert.rejects(execCtl(`instance save transfer 77 transfer.zip ${pri}`));
+						});
+						it("should fail if target instance is not assigned", async function() {
+							await uploadSave(pri, "transfer.zip", "transfer.zip");
+							await assert.rejects(execCtl(`instance save transfer ${pri} transfer.zip 77`));
+							await deleteSave(pri, "transfer.zip");
+						});
+					}
+				});
+			}
+
+		});
+
 		describe("instance save delete", function() {
 			it("should delete a save", async function() {
 				await execCtl("instance save delete 44 upload.zip");