Merge pull request #1114 from ILLISIS/ps-prefetch

Send player data ahead (prefetch) to the instance a player is switching to

plugins/player_sync/README.md

@@ -91,6 +91,25 @@ Communication between the instance and the controller goes over websockets.
 Overall, we are able to achieve a latency between 3 and infinite ticks from server join, depending on the size of the inventory. The major limiter is rcon transfer speeds with larger inventories, especially if they contain blueprints.
 
 
+## Sending player data ahead
+
+A plugin that moves a player to another instance with `LuaPlayer.connect_to_server` can tell player sync where the player is going, just before the call:
+
+```lua
+local ps = remote.interfaces.player_sync
+if ps and ps.set_destination then
+	pcall(remote.call, "player_sync", "set_destination", player.name, instance_id) -- a failure never stops the connect
+end
+player.connect_to_server({ address = address, name = name })
+```
+
+`instance_id` is the Clusterio instance id of the destination. The call only records it, and a later call replaces it.
+When the player then leaves with the reason `switching_servers`, the upload carries the destination and the controller sends the stored data on to that instance while the client reconnects.
+On joining, a copy that the acquire reply confirms is applied without a download.
+The destination's `player_sync.sync_*` options apply to the copy as to a download.
+A cancelled dialog, a quit or a crash sends nothing.
+If the destination isn't running, has player sync off or is the player's current instance, the join downloads as before.
+
 ## Controller Configuration
 
 ### player_sync.player_lock_timeout

plugins/player_sync/controller.ts

@@ -2,6 +2,7 @@ import type { ControllerPluginContext, InstanceRecord } from "@clusterio/control
 import type { IpcPlayerData } from "./messages.js";
 import { mergePlayerData } from "./components.js";
 
+import crypto from "node:crypto";
 import fs from "node:fs/promises";
 import path from "path";
 import * as lib from "@clusterio/lib";
@@ -67,6 +68,8 @@ async function saveDatabase(
 
 export default async function(context: ControllerPluginContext) {
 	const { controller, logger, plugin } = context;
+	// Differs on every controller start: a prefetched copy holding another run's token may be one this run never saved
+	const runToken = crypto.randomBytes(8).toString("hex");
 	const acquiredPlayers = new Map<string, { instanceId: number, expiresMs?: number }>();
 	const { playerDatastore, tombstones } = await loadDatabase(controller.config, logger);
 	let playerDatastoreDirty = false;
@@ -113,6 +116,26 @@ export default async function(context: ControllerPluginContext) {
 		}
 	}
 
+	// Send stored player data on to the instance the player left to switch to (set_destination in the Lua module).
+	// Only another running instance with player_sync on: a copy applied on the source would rebuild the body there.
+	function prefetch(sourceId: number, targetId: number, playerData: IpcPlayerData) {
+		const target = controller.instances.get(targetId);
+		if (
+			!target || targetId === sourceId || target.status !== "running"
+			|| !target.config.get("player_sync.load_plugin")
+		) {
+			logger.verbose(`Not prefetching ${playerData.name} to instance ${targetId}`);
+			return;
+		}
+		try {
+			controller.sendTo({ instanceId: targetId }, new msg.PrefetchEvent(playerData, runToken));
+			logger.verbose(`Prefetching ${playerData.name} to ${target.config.get("instance.name")}`);
+		} catch (err: any) {
+			// sendEvent throws when the instance's host isn't connected; the upload is stored either way
+			logger.warn(`Failed to prefetch ${playerData.name} to instance ${targetId}: ${err.message}`);
+		}
+	}
+
 	controller.hooks.instanceStatusChanged.attach(plugin.name, async (instance: InstanceRecord) => {
 		let instanceId = instance.id;
 		if (["unassigned", "deleted"].includes(instance.status)) {
@@ -166,6 +189,7 @@ export default async function(context: ControllerPluginContext) {
 			playerDatastore.has(playerName),
 			undefined,
 			epoch || undefined,
+			runToken,
 		);
 	});
 
@@ -225,9 +249,14 @@ export default async function(context: ControllerPluginContext) {
 
 		if (store) {
 			logger.verbose(`Received player data for ${playerName} from ${instanceName}`);
-			playerDatastore.set(playerName, mergePlayerData(playerDatastore.get(playerName), playerData));
+			const merged = mergePlayerData(playerDatastore.get(playerName), playerData);
+			playerDatastore.set(playerName, merged);
 			tombstones.delete(playerName);
 			playerDatastoreDirty = true;
+			if (request.destinationInstanceId !== undefined) {
+				// The stored copy: components the upload left out are the controller's
+				prefetch(instanceId, request.destinationInstanceId, merged);
+			}
 		}
 	});
 

plugins/player_sync/index.ts

@@ -114,6 +114,7 @@ export const plugin: lib.PluginDeclaration = {
 		messages.ReleaseRequest,
 		messages.UploadRequest,
 		messages.DownloadRequest,
+		messages.PrefetchEvent,
 		messages.DatabaseStatsRequest,
 		messages.ListPlayersRequest,
 		messages.GetPlayerDataRequest,

plugins/player_sync/instance.ts

@@ -4,6 +4,7 @@ import * as lib from "@clusterio/lib";
 import type { InstancePluginContext } from "@clusterio/host";
 import {
 	AcquireRequest, AcquireResponse, ReleaseRequest, UploadRequest, DownloadRequest, DownloadResponse, IpcPlayerData,
+	PrefetchEvent,
 } from "./messages.js";
 import { type Component, enabledComponents, stripPlayerData } from "./components.js";
 
@@ -28,6 +29,7 @@ type IpcAcquireResponse = {
 	has_data?: boolean,
 	message?: string,
 	epoch?: number,
+	token?: string,
 }
 
 /**
@@ -60,6 +62,8 @@ async function encodeLuaString(text: string): Promise<string> {
 export type RecipeNotificationDelta = {
 	add?: string[],
 	remove?: string[],
+	/** add is the whole stored list, sent without the instance's: the save works out what to remove */
+	full?: true,
 }
 
 /**
@@ -86,6 +90,7 @@ export function recipeNotificationDelta(stored: string[], current: string[]): Re
 /**
  * Replace the stored recipe notifications with the difference from the
  * current state on the instance to reduce the amount of data sent to Lua.
+ * Without a current state the whole stored list is sent, marked full.
  */
 export async function applyRecipeNotificationDelta(
 	playerData: IpcPlayerData,
@@ -96,10 +101,10 @@ export async function applyRecipeNotificationDelta(
 		return;
 	}
 	try {
-		const delta = recipeNotificationDelta(
-			JSON.parse(await decodeLuaString(playerData.recipe_notifications)),
-			current ? JSON.parse(await decodeLuaString(current)) : [],
-		);
+		const stored = JSON.parse(await decodeLuaString(playerData.recipe_notifications));
+		const delta: RecipeNotificationDelta = current
+			? recipeNotificationDelta(stored, JSON.parse(await decodeLuaString(current)))
+			: { ...recipeNotificationDelta(stored, []), full: true };
 		playerData.recipe_notifications = await encodeLuaString(JSON.stringify(delta));
 	} catch (err: any) {
 		logger.warn(`Dropping invalid recipe notifications for ${playerData.name}:\n${err.stack}`);
@@ -135,6 +140,7 @@ export default async function(context: InstancePluginContext) {
 					has_data: acquireResponse.hasData,
 					message: acquireResponse.message,
 					epoch: acquireResponse.epoch,
+					token: acquireResponse.token,
 				};
 			} catch (err: any) {
 				if (!(err instanceof lib.SessionLost)) {
@@ -172,11 +178,16 @@ export default async function(context: InstancePluginContext) {
 			return;
 		}
 
+		// Set by the Lua module when the player left to switch servers (set_destination, which only records integers).
+		// It isn't player data.
+		const destinationInstanceId = player_data.destination_instance_id;
+		delete player_data.destination_instance_id;
+
 		logger.verbose(`Uploading ${player_data.name} (${JSON.stringify(player_data).length / 1000}kB)`);
 		try {
 			await instance.sendTo(
 				"controller",
-				new UploadRequest(instance.id, player_data.name, player_data),
+				new UploadRequest(instance.id, player_data.name, player_data, destinationInstanceId),
 			);
 
 		} catch (err: any) {
@@ -191,6 +202,39 @@ export default async function(context: InstancePluginContext) {
 		);
 	}
 
+	/**
+	 * Send player data to the save in chunks of player_sync.rcon_chunk_size, stripped to the components synced here
+	 * and with its recipe notifications as a delta from current. command(chunk, number, total, extra) gives the RCON
+	 * command for each escaped chunk, numbered from 1; extra passes the components the data was stripped by with
+	 * chunk 1 and is empty for the rest. With stopOnReply a non-empty reply (the save refusing or ignoring the data)
+	 * ends the transfer; without it any reply throws, as sendRcon's expectEmpty does.
+	 */
+	async function sendPlayerData(
+		playerData: IpcPlayerData,
+		current: string | undefined,
+		command: (chunk: string, number: number, total: number, extra: string) => string,
+		stopOnReply: boolean,
+	) {
+		const enabled = enabledComponents(instance.config);
+		stripPlayerData(playerData, enabled);
+		await applyRecipeNotificationDelta(playerData, current, logger);
+
+		const chunks = chunkify(instance.config.get("player_sync.rcon_chunk_size"), JSON.stringify(playerData));
+		logger.verbose(`Sending inventory for ${playerData.name} in ${chunks.length} chunks`);
+		// The scenario restarts a download, and drops a prefetched copy at the join, sent with other components
+		const components = componentsJson(enabled);
+		for (let i = 0; i < chunks.length; i++) {
+			const extra = i === 0 ? `,'${components}'` : "";
+			const reply = await instance.sendRcon(
+				command(lib.escapeString(chunks[i]), i + 1, chunks.length, extra), !stopOnReply, plugin.name,
+			);
+			if (stopOnReply && reply !== "") {
+				logger.verbose(`Stopped sending ${playerData.name}: ${reply.trim()}`);
+				return;
+			}
+		}
+	}
+
 	async function handleDownload(request: IpcDownloadRequest) {
 		const playerName = request.player_name;
 		logger.verbose(`Downloading ${playerName}`);
@@ -207,24 +251,9 @@ export default async function(context: InstancePluginContext) {
 			return;
 		}
 
-		const enabled = enabledComponents(instance.config);
-		stripPlayerData(response.playerData, enabled);
-		await applyRecipeNotificationDelta(response.playerData, request.recipe_notifications, logger);
-
-		const chunkSize = instance.config.get("player_sync.rcon_chunk_size");
-		const chunks = chunkify(chunkSize, JSON.stringify(response.playerData));
-		logger.verbose(`Sending inventory for ${playerName} in ${chunks.length} chunks`);
-		// The scenario restarts the download if it was started with other components
-		const components = componentsJson(enabled);
-		for (let i = 0; i < chunks.length; i++) {
-			const chunk = lib.escapeString(chunks[i]);
-			const extra = i === 0 ? `,'${components}'` : "";
-			await instance.sendRcon(
-				`/sc player_sync.download_inventory('${playerName}','${chunk}',${i + 1},${chunks.length}${extra})`,
-				true,
-				plugin.name,
-			);
-		}
+		await sendPlayerData(response.playerData, request.recipe_notifications, (chunk, number, total, extra) => (
+			`/sc player_sync.download_inventory('${playerName}','${chunk}',${number},${total}${extra})`
+		), false);
 	}
 
 	// Handle IPC from scenario script
@@ -253,6 +282,22 @@ export default async function(context: InstancePluginContext) {
 		),
 	);
 
+	// A copy of a player's stored data, sent ahead while they switch to this instance (player_sync.prefetch_inventory)
+	async function handlePrefetch(event: PrefetchEvent) {
+		const { playerData, token } = event;
+		const { name, generation } = playerData;
+		const epoch = playerData.epoch ?? 0;
+		logger.verbose(`Prefetching ${name}`);
+		// This side doesn't know the save's cleared recipe list, so the whole stored list goes as a full delta
+		await sendPlayerData(playerData, undefined, (chunk, number, total, extra) => {
+			// Later chunks have no components argument, so nil holds its place
+			const copyArgs = number === 1 ? `,'${token}',${epoch}` : `,nil,'${token}',${epoch}`;
+			const args = `'${name}',${generation},'${chunk}',${number},${total}${extra}${copyArgs}`;
+			return `/sc player_sync.prefetch_inventory(${args})`;
+		}, true);
+	}
+	instance.handle(PrefetchEvent, handlePrefetch);
+
 	/** Tell the scenario which components are synced. */
 	async function sendComponents() {
 		const json = componentsJson(enabledComponents(instance.config));

plugins/player_sync/messages.ts

@@ -29,6 +29,7 @@ export type IpcPlayerData = {
 	recipe_notifications?: string,
 	// Components the upload was serialized with, all when missing
 	components?: Record<string, boolean> | unknown[],
+	destination_instance_id?: number, // set_destination's instance, on the leave upload only; not player data
 }
 
 // Marks tombstones in player_sync.json
@@ -195,6 +196,7 @@ export class AcquireResponse {
 		public hasData?: boolean,
 		public message?: string,
 		public epoch?: number,
+		public token?: string,
 	) {
 	}
 
@@ -204,10 +206,11 @@ export class AcquireResponse {
 		"hasData": Type.Optional(Type.Boolean()),
 		"message": Type.Optional(Type.String()),
 		"epoch": Type.Optional(Type.Integer()),
+		"token": Type.Optional(Type.String()),
 	});
 
 	static fromJSON(json: Static<typeof AcquireResponse.jsonSchema>): AcquireResponse {
-		return new this(json.status, json.generation, json.hasData, json.message, json.epoch);
+		return new this(json.status, json.generation, json.hasData, json.message, json.epoch, json.token);
 	}
 };
 export class AcquireRequest {
@@ -268,6 +271,7 @@ export class UploadRequest {
 		public instanceId: number,
 		public playerName: string,
 		public playerData: IpcPlayerData,
+		public destinationInstanceId?: number,
 	) {
 	}
 
@@ -275,10 +279,36 @@ export class UploadRequest {
 		"instanceId": Type.Integer(),
 		"playerName": Type.String(),
 		"playerData": jsonPlayerData,
+		"destinationInstanceId": Type.Optional(Type.Integer()),
 	});
 
 	static fromJSON(json: Static<typeof UploadRequest.jsonSchema>) {
-		return new this(json.instanceId, json.playerName, json.playerData as IpcPlayerData);
+		return new this(json.instanceId, json.playerName, json.playerData as IpcPlayerData, json.destinationInstanceId);
+	}
+}
+
+/** The stored data of a player switching to the instance it is sent to, so the join there needs no download */
+export class PrefetchEvent {
+	declare ["constructor"]: typeof PrefetchEvent;
+	static type = "event" as const;
+	static src = "controller" as const;
+	static dst = "instance" as const;
+	static plugin = "player_sync" as const;
+
+	constructor(
+		public playerData: IpcPlayerData,
+		/** The controller run that sent it, which a held copy must share with the acquire reply to be applied */
+		public token: string,
+	) {
+	}
+
+	static jsonSchema = Type.Object({
+		"playerData": jsonPlayerData,
+		"token": Type.String(),
+	});
+
+	static fromJSON(json: Static<typeof PrefetchEvent.jsonSchema>): PrefetchEvent {
+		return new this(json.playerData as IpcPlayerData, json.token);
 	}
 }
 

plugins/player_sync/module/download_inventory.lua

@@ -8,18 +8,22 @@ local compat = require("modules/clusterio/compat")
 
 --- @param components string? JSON of the components the instance sent the data with, given with the first part
 local function download_inventory(player_name, data, number, total, components)
+	local script_data = get_script_data()
+	local record = script_data.active_downloads[player_name]
 	local player = game.get_player(player_name)
-	if player == nil then
+	-- A prefetched copy (player_sync.prefetch_inventory) may be for a player who doesn't exist here yet
+	if player == nil and not (record and record.prefetch) then
 		rcon.print("Player " .. player_name .. " does not exist")
 		return
 	end
 
-	local script_data = get_script_data()
-	local record = script_data.active_downloads[player_name]
 	if record == nil then
 		rcon.print("No active download is in progress for " .. player_name)
 		return
 	end
+	-- For the linter, which can't narrow through the check above. A prefetch record never reaches a line that indexes
+	-- player while it is nil: it has no restart, never gets total 0, and shown below is false for it.
+	--- @cast player LuaPlayer
 
 	local player_record = script_data.players[player_name]
 	-- The components changed after the download was started
@@ -60,30 +64,40 @@ local function download_inventory(player_name, data, number, total, components)
 	end
 
 	record.data = record.data .. data
+	-- A prefetched copy shows progress, and is applied, only once the acquire reply has confirmed it for the player
+	-- online here (player_sync.sync_player). A copy that finishes before then waits in finished_downloads for that reply.
+	local shown = not record.prefetch or (record.confirmed and player.connected)
 
 	-- Show progress in console
 	-- player.print("Downloaded "..(number + 1).."/"..total.." parts")
 	if number ~= total then
 		-- Show progress in GUI
-		progress_dialog.display(player, number, total)
+		if shown then
+			progress_dialog.display(player, number, total)
+		end
 
 		-- Update activity to indicate this session is still going
 		record.last_active = game.ticks_played
 		return
 	end
 
-	-- Download is complete
-	progress_dialog.remove(player)
+	-- Download is complete. The dialog goes even when this copy isn't shown now: it may have been shown before a leave
+	if player then
+		progress_dialog.remove(player)
+	end
 
 	-- Remove download session and store finished download
 	script_data.active_downloads[player_name] = nil
 	script_data.finished_downloads[player_name] = record
 
 	local ticks = game.ticks_played - record.started
-	player.print("Imported player data in " .. ticks .. " ticks")
-	log("Imported player data for " .. player_name .. " in " .. ticks .. " ticks")
+	if shown then
+		player.print("Imported player data in " .. ticks .. " ticks")
+	end
+	local kind = record.prefetch and "Prefetched" or "Imported"
+	log(kind .. " player data for " .. player_name .. " in " .. ticks .. " ticks")
 
-	if player.connected then
+	if shown and player.connected then
 		player_sync.finish_download(player, record)
 	end
 end

plugins/player_sync/module/player_sync.lua

@@ -73,6 +73,36 @@ local function remove_inventory_bonus_raise(script_data, player_name)
 	end
 end
 
+-- Turn a synced player into a spectator while their player data is downloaded or applied, if this instance syncs
+-- inventories, and keep where their body stood in record for restore_position. The old body goes, with its crafting
+-- queue. Until the data is applied, the player's state isn't persisted.
+local function spectate(player, player_record, record)
+	-- Indicate the current state shouldn't be persisted
+	player_record.sync_in_progress = true
+
+	if player_record.sync and serialize.syncs(record.components, "inventories") then
+		-- Store original position to teleport back to. Only once per record: a rejoin while a prefetched copy is held
+		-- spectates again, and the spectator's position then isn't where the body stood
+		if record.position == nil then
+			if v2_remote_controller then
+				record.surface = player.physical_surface
+				record.position = player.physical_position
+			else
+				record.surface = player.surface
+				record.position = player.position
+			end
+			if player.driving then
+				record.vehicle = player.vehicle
+			end
+		end
+
+		save_local_controller(player, player_record, record.components)
+		local character = player.character
+		player.set_controller({ type = defines.controllers.spectator })
+		if character ~= nil then character.destroy() end
+	end
+end
+
 
 player_sync = {}
 -- This function is called by instance.js as /sc player_sync.download_inventory("danielv", Escaped JSON string, package_number, total_packages_count) with data from the controller
@@ -83,6 +113,24 @@ function player_sync.set_components(data)
 	get_script_data().components = assert(compat.json_to_table(data))
 end
 
+-- For plugins that move players to another instance: remote.call("player_sync", "set_destination", player_name,
+-- instance_id) just before LuaPlayer.connect_to_server, with the destination's Clusterio instance id. It only records
+-- it; if the player then leaves to switch servers, the upload carries it and the controller sends the player's data
+-- on to that instance (README, "Sending player data ahead"). A later call replaces it, and every leave clears it.
+remote.add_interface("player_sync", {
+	set_destination = function(player_name, instance_id)
+		-- Integers only: table_to_json writes inf and nan bare, which the host can't parse, so the upload carrying it
+		-- would fail on every retry and the session would be lost (inf % 1 and nan % 1 are nan, never 0)
+		if type(instance_id) ~= "number" or instance_id % 1 ~= 0 then
+			return
+		end
+		local player_record = get_script_data().players[player_name]
+		if player_record then
+			player_record.destination = instance_id
+		end
+	end,
+})
+
 -- This function is called internally when a player leaves the game to
 -- serialize the player for upload.
 function player_sync.serialize_player(player, player_record)
@@ -286,6 +334,19 @@ function player_sync.check_players_in_cutscene()
 	end
 end
 
+-- Whether a join should take the stored data: there is some, and this save hasn't synced the player in that epoch at
+-- that generation or newer with every component synced here now. A player without a record here has nothing to keep.
+local function download_needed(player_record, acquire_response, script_data)
+	return acquire_response.has_data
+		and (
+			not player_record
+			or not player_record.sync
+			or (player_record.epoch or 0) ~= (acquire_response.epoch or 0)
+			or acquire_response.generation > player_record.generation
+			or serialize.newly_synced(player_record.components, script_data.components)
+		)
+end
+
 function player_sync.sync_player(acquire_response)
 	local player = assert(game.get_player(acquire_response.player_name))
 
@@ -303,6 +364,31 @@ function player_sync.sync_player(acquire_response)
 	local script_data = get_script_data()
 	-- Epoch is bumped when an admin replaces or deletes the stored data
 	local epoch = acquire_response.epoch or 0
+	-- A prefetched copy (player_sync.prefetch_inventory) is used only if this reply acquires the player at the copy's
+	-- generation and epoch, from the controller run that sent it (token), and the copy was sent with the components
+	-- synced here now. The generation alone isn't enough: a controller that crashed before saving it restarts at the
+	-- last saved generation, and another instance's upload can then reuse the same number with other data. Otherwise
+	-- the copy goes, and the join carries on as if it had never come: today's download or failure dialog, never the
+	-- restart path below, which only sees a different generation or epoch. A copy this save doesn't need goes too,
+	-- keeping the local data: an upload from an instance with older data can take the generation this save already
+	-- holds.
+	-- A copy confirmed at an earlier join already took the body (spectate keeps where it stood), and the local data
+	-- went with it, so a join that drops that copy downloads.
+	local body_taken -- the dropped copy that took the body
+	for _, downloads in ipairs({ script_data.finished_downloads, script_data.active_downloads }) do
+		local record = downloads[acquire_response.player_name]
+		if record and record.prefetch and (not (
+			acquire_response.status == "acquired" and record.generation == acquire_response.generation
+			and (record.epoch or 0) == epoch and record.token == acquire_response.token
+			and serialize.same_components(record.components, script_data.components)
+		) or not download_needed(script_data.players[acquire_response.player_name], acquire_response, script_data)) then
+			if record.position ~= nil then
+				body_taken = record
+			end
+			downloads[acquire_response.player_name] = nil
+		end
+	end
+
 	local finished_record = script_data.finished_downloads[acquire_response.player_name]
 	if finished_record then
 		if
@@ -310,6 +396,14 @@ function player_sync.sync_player(acquire_response)
 			and finished_record.generation == acquire_response.generation
 			and (finished_record.epoch or 0) == epoch
 		then
+			if finished_record.prefetch then
+				-- As a download does: the body left here goes, and the copy makes a new character where it stood
+				spectate(player, script_data.players[player.name], finished_record)
+				if player_sync.finish_download(player, finished_record) then
+					log("Applied prefetched player data for " .. player.name)
+				end
+				return
+			end
 			player_sync.finish_download(player, finished_record)
 			return
 		end
@@ -327,6 +421,13 @@ function player_sync.sync_player(acquire_response)
 			and download_record.generation == acquire_response.generation
 			and (download_record.epoch or 0) == epoch
 		then
+			if download_record.prefetch then
+				-- The copy is still arriving: spectate as a download does, and its last chunk applies it
+				spectate(player, script_data.players[player.name], download_record)
+				progress_dialog.display(player, 0, 1)
+				download_record.confirmed = true
+				return
+			end
 			game.print("Continuing active download")
 			return
 		end
@@ -378,16 +479,15 @@ function player_sync.sync_player(acquire_response)
 	end
 
 	local stale = (player_record.epoch or 0) ~= epoch
-	if
-		acquire_response.has_data
-		and (
-			not player_record.sync
-			or stale
-			or acquire_response.generation > player_record.generation
-			or serialize.newly_synced(player_record.components, script_data.components)
-		)
-	then
+	if body_taken or download_needed(player_record, acquire_response, script_data) then
 		player_sync.initiate_inventory_download(player, player_record, acquire_response.generation, epoch)
+		if body_taken then
+			-- spectate kept the spectator's place, the body stood where the dropped copy kept
+			local download = script_data.active_downloads[player.name]
+			download.surface = body_taken.surface
+			download.position = body_taken.position
+			download.vehicle = body_taken.vehicle
+		end
 
 	else
 		if acquire_response.generation == 0 then
@@ -412,6 +512,7 @@ function player_sync.sync_player(acquire_response)
 	end
 end
 
+--- @return boolean? applied true when the player data was deserialized
 function player_sync.finish_download(player, finished_record)
 	local script_data = get_script_data()
 	script_data.finished_downloads[player.name] = nil
@@ -433,6 +534,7 @@ function player_sync.finish_download(player, finished_record)
 	start_sync(player_record, finished_record.components)
 	player_record.generation = finished_record.generation
 	player_record.epoch = finished_record.epoch
+	return true
 end
 
 -- Download inventory from controller
@@ -486,9 +588,19 @@ player_sync.events[defines.events.on_pre_player_left_game] = function(event)
 	local script_data = get_script_data()
 	local player = assert(game.get_player(event.player_index))
 	local player_record = script_data.players[player.name]
+	-- Recorded by set_destination. Every leave clears it; only an uploading switching_servers leave sends it on.
+	local destination = player_record.destination
+	player_record.destination = nil
 
-	-- If player has an active download release our acquisition but let the download continue
-	if script_data.active_downloads[player.name] then
+	local download = script_data.active_downloads[player.name]
+	if download and download.prefetch then
+		-- A prefetched copy is confirmed per join: the next join's acquire reply confirms it again. Until then it has
+		-- replaced nothing here, so the checks below decide the upload. One whose join took the body (spectate) left
+		-- sync_in_progress set, and uploads nothing.
+		download.confirmed = nil
+
+	elseif download then
+		-- If player has an active download release our acquisition but let the download continue
 		clusterio_api.send_json("player_sync_release", { player_name = player.name })
 		return
 	end
@@ -506,6 +618,11 @@ player_sync.events[defines.events.on_pre_player_left_game] = function(event)
 		player.print("ERROR: Serializing player data failed: " .. result)
 		return
 	end
+	-- The controller sends the stored data on to that instance while the client reconnects. In the serialized table,
+	-- so a retried upload (check_active_uploads) carries it too.
+	if destination and event.reason == defines.disconnect_reason.switching_servers then
+		result.destination_instance_id = destination
+	end
 	script_data.active_uploads[player.name] = {
 		serialized = result,
 		last_attempt = game.ticks_played,
@@ -563,29 +680,7 @@ function player_sync.initiate_inventory_download(player, player_record, generati
 		recipe_notifications = recipe_notifications,
 	})
 
-	-- Indicate the current state shouldn't be persisted
-	player_record.sync_in_progress = true
-
-	-- If this is a synced player turn them into a spectator while the
-	-- player data is downloading
-	if player_record.sync and serialize.syncs(record.components, "inventories") then
-		-- Store original position to teleport back to
-		if v2_remote_controller then
-			record.surface = player.physical_surface
-			record.position = player.physical_position
-		else
-			record.surface = player.surface
-			record.position = player.position
-		end
-		if player.driving then
-			record.vehicle = player.vehicle
-		end
-
-		save_local_controller(player, player_record, record.components)
-		local character = player.character
-		player.set_controller({ type = defines.controllers.spectator })
-		if character ~= nil then character.destroy() end
-	end
+	spectate(player, player_record, record)
 
 	progress_dialog.display(player, 0, 1)
 end
@@ -632,4 +727,73 @@ function player_sync.check_active_uploads()
 	end
 end
 
+-- Called by instance.ts as /sc player_sync.prefetch_inventory('name', generation, 'chunk', number, total, components,
+-- token, epoch) with a copy of a player's stored data that the controller sent ahead (PrefetchEvent) while they switch
+-- to this server. It holds the copy for a player who is offline here or doesn't exist yet. Every chunk carries the
+-- token and epoch, so chunks of another copy never land in this one. It prints only when it refuses or ignores a chunk,
+-- which stops the host sending the rest.
+--- @param components string? JSON of the components the instance stripped the copy by, given with the first chunk
+--- @param token string? The controller run that sent the copy
+--- @param epoch integer? The stored data's epoch, bumped when an admin replaces or deletes it
+function player_sync.prefetch_inventory(player_name, generation, data, number, total, components, token, epoch)
+	epoch = epoch or 0
+	local script_data = get_script_data()
+	if number == 1 then
+		local player = game.get_player(player_name)
+		local active = script_data.active_downloads[player_name]
+		local finished = script_data.finished_downloads[player_name]
+		if player and player.connected then
+			rcon.print("Refused prefetch for " .. player_name .. ": online here, their join downloads")
+			return
+		end
+		if active and not active.prefetch then
+			-- download_inventory has no generation check, so its remaining chunks would land in the copy
+			rcon.print("Refused prefetch for " .. player_name .. ": a download is in progress")
+			return
+		end
+		-- Two copies can reach here out of order (applyRecipeNotificationDelta awaits zlib). A held finished copy of the
+		-- same generation is replaced: the newer copy of the same number is what the controller holds now. A copy
+		-- with another token came from an earlier controller run, which may have crashed before saving that
+		-- generation, and one with another epoch from the other side of an admin's import or delete, which can start
+		-- the generations over. Either way the generations don't compare, and the arriving copy replaces the held one
+		local function comparable(held)
+			return held.token == token and (held.epoch or 0) == epoch
+		end
+		local active_newer = active and comparable(active) and active.generation >= generation
+		local finished_newer = finished and comparable(finished) and finished.generation > generation
+		if active_newer or finished_newer then
+			rcon.print("Refused prefetch for " .. player_name .. ": generation " .. generation .. " is not newer")
+			return
+		end
+		-- A held copy (or offline-finished download) whose join took the body knows where it stood, and that the body
+		-- is gone (sync_player's body_taken): the copy replacing it keeps both
+		local held = active or finished
+		script_data.finished_downloads[player_name] = nil
+		script_data.active_downloads[player_name] = {
+			started = game.ticks_played,
+			last_active = game.ticks_played,
+			generation = generation,
+			data = "",
+			prefetch = true,
+			token = token,
+			epoch = epoch,
+			components = components and compat.json_to_table(components),
+			surface = held and held.surface,
+			position = held and held.position,
+			vehicle = held and held.vehicle,
+		}
+	end
+
+	local record = script_data.active_downloads[player_name]
+	if not (
+		record and record.prefetch and record.generation == generation and record.token == token
+		and (record.epoch or 0) == epoch
+	) then
+		-- Refused at chunk 1, dropped by the acquire reply or the inactivity check, or replaced by a newer copy
+		rcon.print("Ignored prefetch chunk " .. number .. " for " .. player_name)
+		return
+	end
+	download_inventory(player_name, data, number, total)
+end
+
 return player_sync

plugins/player_sync/module/serialize.lua

@@ -895,6 +895,7 @@ end
 --- @class CraftingNotificationDelta
 --- @field add string[]? Recipes cleared elsewhere that still have a notification here
 --- @field remove string[]? Recipes with a notification elsewhere that are cleared here
+--- @field full boolean? add is the whole stored list, sent without this save's (a prefetched copy)
 
 --- @param player LuaPlayer
 --- @param serialized string Encoded delta against the state sent with the download request
@@ -929,6 +930,23 @@ function serialize.deserialize_crafting_notifications(player, serialized)
 		end
 	end
 
+	-- A full delta has no remove: recipes cleared here that aren't in add get their notification back, as remove does
+	if delta.full then
+		local stored = {}
+		for _, recipe_name in pairs(delta.add or {}) do
+			stored[recipe_name] = true
+		end
+		local notified = {}
+		for _, recipe in pairs(player.get_recipe_notifications()) do
+			notified[recipe.name] = true
+		end
+		for recipe_name in pairs(enabled) do
+			if not stored[recipe_name] and not notified[recipe_name] then
+				player.add_recipe_notification(recipe_name)
+			end
+		end
+	end
+
 	return next(failed) and failed or nil
 end
 
@@ -959,6 +977,7 @@ end
 --- @field personal_logistic_slots table?
 --- @field crafting_queue table?
 --- @field recipe_notifications string?
+--- @field destination_instance_id integer? Only on a leave upload, for instance.ts, which strips it
 
 --- @param player LuaPlayer
 --- @return table<string, boolean>

plugins/player_sync/test/controller.js

@@ -112,7 +112,10 @@ describe("player_sync controller", function() {
 			await handler(msg.UploadRequest)(new msg.UploadRequest(instanceId, data.name, data));
 		}
 		async function acquireResponse(instanceId, playerName) {
-			return await handler(msg.AcquireRequest)(new msg.AcquireRequest(instanceId, playerName));
+			const response = await handler(msg.AcquireRequest)(new msg.AcquireRequest(instanceId, playerName));
+			// Random per controller run, the run token tests cover it
+			response.token = undefined;
+			return response;
 		}
 
 		async function loadWithInstances() {
@@ -386,4 +389,94 @@ describe("player_sync controller", function() {
 			});
 		});
 	});
+
+	describe("UploadRequest with destinationInstanceId", function() {
+		// Instances 1-3 run player_sync, 4 is stopped, 5 runs with player_sync off; 1 is the source in every case
+		async function setup() {
+			const controller = await load();
+			for (const [id, status, loadPlugin] of [
+				[1, "running", true], [2, "running", true], [3, "running", true],
+				[4, "stopped", true], [5, "running", false],
+			]) {
+				const fields = { "instance.name": `instance ${id}`, "player_sync.load_plugin": loadPlugin };
+				controller.instances.records.set(id, { id, status, config: { get: field => fields[field] } });
+			}
+			const sent = [];
+			controller.sendTo = (address, message) => { sent.push({ address, message }); };
+			return { controller, sent, upload: mock.getHandler(controller, msg.UploadRequest) };
+		}
+		const data = generation => ({ name: "test", generation });
+		async function stored(controller) {
+			const download = mock.getHandler(controller, msg.DownloadRequest);
+			return (await download(new msg.DownloadRequest(2, "test"))).playerData;
+		}
+
+		it("should send the stored data to a running destination with player_sync on", async function() {
+			const { sent, upload } = await setup();
+			await upload(new msg.UploadRequest(1, "test", data(1), 2));
+			assert.equal(sent.length, 1);
+			assert.deepEqual(sent[0].address, { instanceId: 2 });
+			assert.ok(sent[0].message instanceof msg.PrefetchEvent);
+			assert.deepEqual(sent[0].message.playerData, data(1));
+		});
+		it("should tag the prefetch and acquire reply with one token per controller run", async function() {
+			const { controller, sent, upload } = await setup();
+			await upload(new msg.UploadRequest(1, "test", data(1), 2));
+			const acquire = await mock.getHandler(controller, msg.AcquireRequest)(new msg.AcquireRequest(2, "test"));
+			assert.equal(acquire.status, "acquired");
+			assert.ok(sent[0].message.token);
+			assert.equal(acquire.token, sent[0].message.token);
+
+			const again = await setup();
+			await again.upload(new msg.UploadRequest(1, "test", data(1), 2));
+			assert.notEqual(again.sent[0].message.token, sent[0].message.token);
+		});
+		it("should not need a token on a busy reply", async function() {
+			const { controller } = await setup();
+			const acquire = mock.getHandler(controller, msg.AcquireRequest);
+			await acquire(new msg.AcquireRequest(3, "test"));
+			assert.equal((await acquire(new msg.AcquireRequest(2, "test"))).token, undefined);
+		});
+		it("should send the stored copy, with the parts the upload's components left out", async function() {
+			const { sent, upload } = await setup();
+			await upload(new msg.UploadRequest(1, "test", { ...data(1), force: "player", tag: "first" }));
+			const components = Object.fromEntries(Object.keys(allComponents).map(c => [c, c !== "force"]));
+			await upload(new msg.UploadRequest(1, "test", { ...data(2), tag: "second", components }, 2));
+			assert.equal(sent.length, 1);
+			assert.deepEqual(sent[0].message.playerData, { ...data(2), force: "player", tag: "second" });
+		});
+		it("should send nothing without a destination", async function() {
+			const { sent, upload } = await setup();
+			await upload(new msg.UploadRequest(1, "test", data(1)));
+			assert.equal(sent.length, 0);
+		});
+		it("should send nothing when an older generation isn't stored", async function() {
+			const { sent, upload } = await setup();
+			await upload(new msg.UploadRequest(1, "test", data(3)));
+			await upload(new msg.UploadRequest(1, "test", data(2), 2));
+			assert.equal(sent.length, 0);
+		});
+		it("should send nothing when another instance's lock rejects the upload", async function() {
+			const { controller, sent, upload } = await setup();
+			await mock.getHandler(controller, msg.AcquireRequest)(new msg.AcquireRequest(3, "test"));
+			await upload(new msg.UploadRequest(1, "test", data(1), 2));
+			assert.equal(sent.length, 0);
+		});
+		it("should store but not send to a stopped, source, plugin-off or unknown instance", async function() {
+			const { controller, sent, upload } = await setup();
+			let generation = 0;
+			for (const id of [4, 1, 5, 99]) {
+				generation += 1;
+				await upload(new msg.UploadRequest(1, "test", data(generation), id));
+			}
+			assert.equal(sent.length, 0);
+			assert.equal((await stored(controller)).generation, 4);
+		});
+		it("should store the upload and not fail it when sending throws", async function() {
+			const { controller, upload } = await setup();
+			controller.sendTo = () => { throw new Error("Host containing instance is not connected"); };
+			await upload(new msg.UploadRequest(1, "test", data(1), 2));
+			assert.deepEqual(await stored(controller), data(1));
+		});
+	});
 });

plugins/player_sync/test/instance.js

@@ -5,7 +5,7 @@ import * as lib from "@clusterio/lib";
 
 import * as mock from "../../../test/mock.js";
 import loadInstancePlugin, { recipeNotificationDelta, applyRecipeNotificationDelta } from "../dist/node/instance.js";
-import { DownloadResponse } from "../dist/node/messages.js";
+import { AcquireResponse, DownloadResponse, PrefetchEvent } from "../dist/node/messages.js";
 import { components } from "../dist/node/components.js";
 import { plugin as info } from "../dist/node/index.js";
 
@@ -54,10 +54,15 @@ describe("player_sync", function() {
 			await apply(playerData, await encode(["a"]));
 			assert.deepEqual(await decode(playerData.recipe_notifications), {});
 		});
-		it("should add everything without a current list", async function() {
+		it("should send the whole list marked full without a current list", async function() {
 			const playerData = { name: "test", recipe_notifications: await encode(["a"]) };
 			await apply(playerData, undefined);
-			assert.deepEqual(await decode(playerData.recipe_notifications), { add: ["a"] });
+			assert.deepEqual(await decode(playerData.recipe_notifications), { add: ["a"], full: true });
+		});
+		it("should mark an empty list full without a current list", async function() {
+			const playerData = { name: "test", recipe_notifications: await encode([]) };
+			await apply(playerData, undefined);
+			assert.deepEqual(await decode(playerData.recipe_notifications), { full: true });
 		});
 		it("should do nothing without stored notifications", async function() {
 			const playerData = { name: "test" };
@@ -118,10 +123,12 @@ describe("player_sync", function() {
 				["/sc player_sync.download_inventory('test',nil,0,0)"]
 			);
 		});
-		it("should send everything as a delta when no snapshot was given", async function() {
+		it("should send everything as a full delta when no snapshot was given", async function() {
 			respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a"]) });
 			await download({ player_name: "test" });
-			const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) };
+			const expected = {
+				generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"], full: true }),
+			};
 			assert.deepEqual(instance.server.rconCommands, [downloadCommand(expected)]);
 		});
 		it("should send a delta when a snapshot was given", async function() {
@@ -211,4 +218,127 @@ describe("player_sync", function() {
 			assert.equal(instance.server.rconCommands.length, 0);
 		});
 	});
+
+	describe("ipc-player_sync_upload", function() {
+		let instance;
+		before(async function() {
+			try {
+				lib.registerPluginMessages([info]);
+			} catch (err) {
+				// Already registered by the full test suite
+			}
+			({ instance } = await mock.loadInstancePlugin(loadInstancePlugin, info));
+		});
+		beforeEach(function() {
+			instance.connector.sentMessages = [];
+		});
+
+		// The UploadRequest the instance sends the controller (no response is given; only the request is checked)
+		async function upload(playerData) {
+			instance.server.emit("ipc-player_sync_upload", playerData);
+			for (let i = 0; i < 100 && !instance.connector.sentMessages.length; i++) {
+				await lib.wait(1);
+			}
+			return instance.connector.sentMessages[0].data;
+		}
+
+		it("should pass destination_instance_id on as destinationInstanceId, not as player data", async function() {
+			const request = await upload({ name: "test", generation: 2, destination_instance_id: 7 });
+			assert.equal(request.destinationInstanceId, 7);
+			assert.deepEqual(request.playerData, { name: "test", generation: 2 });
+		});
+		it("should send no destination without the field", async function() {
+			const request = await upload({ name: "test", generation: 2 });
+			assert.equal(request.destinationInstanceId, undefined);
+		});
+	});
+
+	describe("PrefetchEvent", function() {
+		let instance;
+		before(async function() {
+			try {
+				lib.registerPluginMessages([info]);
+			} catch (err) {
+				// Already registered by the full test suite
+			}
+			({ instance } = await mock.loadInstancePlugin(loadInstancePlugin, info));
+			instance.mockConfigEntries.set("player_sync.rcon_chunk_size", 10);
+		});
+		beforeEach(function() {
+			instance.server.reset();
+			for (const component of Object.keys(components)) {
+				instance.mockConfigEntries.set(`player_sync.sync_${component}`, true);
+			}
+		});
+
+		const prefetch = playerData => mock.getHandler(instance, PrefetchEvent)(new PrefetchEvent(playerData, "tok"));
+		function prefetchCommands(playerData, enabled = Object.keys(components), epoch = 0) {
+			const chunks = JSON.stringify(playerData).match(/.{1,10}/g);
+			const sent = lib.escapeString(JSON.stringify(Object.fromEntries(enabled.map(c => [c, true]))));
+			return chunks.map((chunk, i) => "/sc player_sync.prefetch_inventory(" +
+				`'test',${playerData.generation},'${lib.escapeString(chunk)}',${i + 1},${chunks.length}` +
+				`${i === 0 ? `,'${sent}','tok',${epoch}` : `,nil,'tok',${epoch}`})`);
+		}
+
+		it("should send prefetch_inventory chunks with the generation and the full recipe list", async function() {
+			await prefetch({ generation: 3, name: "test", recipe_notifications: await encode(["a"]) });
+			const expected = {
+				generation: 3, name: "test", recipe_notifications: await encode({ add: ["a"], full: true }),
+			};
+			assert.ok(instance.server.rconCommands.length > 1, "more than one chunk");
+			assert.deepEqual(instance.server.rconCommands, prefetchCommands(expected));
+		});
+		it("should leave out components which are not synced and send the components with chunk 1", async function() {
+			instance.mockConfigEntries.set("player_sync.sync_force", false);
+			await prefetch({ generation: 3, name: "test", force: "enemy", tag: "tag" });
+			const enabled = Object.keys(components).filter(c => c !== "force");
+			const commands = prefetchCommands({ generation: 3, name: "test", tag: "tag" }, enabled);
+			assert.ok(commands.length > 1, "more than one chunk");
+			assert.deepEqual(instance.server.rconCommands, commands);
+		});
+		it("should send the token and epoch with every chunk", async function() {
+			await prefetch({ generation: 3, name: "test" });
+			const commands = instance.server.rconCommands;
+			assert.ok(commands[0].endsWith(",'tok',0)"));
+			assert.ok(commands.length > 1 && commands.slice(1).every(command => command.endsWith(",nil,'tok',0)")));
+		});
+		it("should send the stored data's epoch", async function() {
+			const playerData = { generation: 3, epoch: 2, name: "test" };
+			await prefetch(playerData);
+			assert.deepEqual(instance.server.rconCommands, prefetchCommands(playerData, undefined, 2));
+		});
+		it("should stop when the save refuses the first chunk", async function() {
+			const playerData = { generation: 3, name: "test" };
+			const commands = prefetchCommands(playerData);
+			instance.server.rconCommandResults.set(commands[0], "Refused prefetch for test: online here");
+			await prefetch(playerData);
+			assert.deepEqual(instance.server.rconCommands, commands.slice(0, 1));
+		});
+		it("should stop at the first non-empty reply mid-stream", async function() {
+			const playerData = { generation: 3, name: "test" };
+			const commands = prefetchCommands(playerData);
+			assert.equal(commands.length, 3);
+			instance.server.rconCommandResults.set(commands[1], "Ignored prefetch chunk 2 for test");
+			await prefetch(playerData);
+			assert.deepEqual(instance.server.rconCommands, commands.slice(0, 2));
+		});
+	});
+
+	describe("acquire reply", function() {
+		it("should pass the controller's token to acquire_response", async function() {
+			try {
+				lib.registerPluginMessages([info]);
+			} catch (err) {
+				// Already registered by the full test suite
+			}
+			const { instance, host } = await mock.loadInstancePlugin(loadInstancePlugin, info);
+			host.connector.connected = true;
+			instance.sendTo = async () => new AcquireResponse("acquired", 4, true, undefined, undefined, "tok");
+			instance.server.emit("ipc-player_sync_acquire", { player_name: "test" });
+			await lib.wait(10);
+			const [command] = instance.server.rconCommands;
+			assert.ok(command.startsWith("/sc player_sync.acquire_response("), command);
+			assert.ok(command.includes(lib.escapeString('"token":"tok"')), command);
+		});
+	});
 });