Tag prefetched copies with the controller's run token

If the controller crashes after sending generation N but before saving it, the destination keeps that prefetch while the controller restarts at N-1, and another instance can then upload different data as generation N. The controller now picks a random token per run, sends it with every prefetch and acquire reply, and the scenario applies a held copy only when its generation and token both match, and lets a new run's copy replace one with another token.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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;
@@ -125,7 +128,7 @@ export default async function(context: ControllerPluginContext) {
 			return;
 		}
 		try {
-			controller.sendTo({ instanceId: targetId }, new msg.PrefetchEvent(playerData));
+			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
@@ -186,6 +189,7 @@ export default async function(context: ControllerPluginContext) {
 			playerDatastore.has(playerName),
 			undefined,
 			epoch || undefined,
+			runToken,
 		);
 	});
 

plugins/player_sync/instance.ts

@@ -29,6 +29,7 @@ type IpcAcquireResponse = {
 	has_data?: boolean,
 	message?: string,
 	epoch?: number,
+	token?: string,
 }
 
 /**
@@ -139,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)) {
@@ -282,13 +284,15 @@ 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 } = event;
+		const { playerData, token } = event;
 		const { name, generation } = playerData;
 		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) => (
-			`/sc player_sync.prefetch_inventory('${name}',${generation},'${chunk}',${number},${total}${extra})`
-		), true);
+		await sendPlayerData(playerData, undefined, (chunk, number, total, extra) => {
+			const tokenArg = number === 1 ? `,'${token}'` : "";
+			const args = `'${name}',${generation},'${chunk}',${number},${total}${extra}${tokenArg}`;
+			return `/sc player_sync.prefetch_inventory(${args})`;
+		}, true);
 	}
 	instance.handle(PrefetchEvent, handlePrefetch);
 

plugins/player_sync/messages.ts

@@ -196,6 +196,7 @@ export class AcquireResponse {
 		public hasData?: boolean,
 		public message?: string,
 		public epoch?: number,
+		public token?: string,
 	) {
 	}
 
@@ -205,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 {
@@ -295,15 +297,18 @@ export class PrefetchEvent {
 
 	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);
+		return new this(json.playerData as IpcPlayerData, json.token);
 	}
 }
 

plugins/player_sync/module/player_sync.lua

@@ -358,6 +358,7 @@ function player_sync.sync_player(acquire_response)
 		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.token == acquire_response.token
 			and serialize.same_components(record.components, script_data.components)
 		) then
 			downloads[acquire_response.player_name] = nil
@@ -698,12 +699,14 @@ function player_sync.check_active_uploads()
 	end
 end
 
--- Called by instance.ts as /sc player_sync.prefetch_inventory('name', generation, 'chunk', number, total) with a copy
--- of a player's stored data that the controller sent ahead (PrefetchEvent) while they switch to this server. It holds
+-- Called by instance.ts as /sc player_sync.prefetch_inventory('name', generation, 'chunk', number, total, components,
+-- token) 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. 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
-function player_sync.prefetch_inventory(player_name, generation, data, number, total, components)
+--- @param token string? The controller run that sent the copy, given with the first chunk
+function player_sync.prefetch_inventory(player_name, generation, data, number, total, components, token)
 	local script_data = get_script_data()
 	if number == 1 then
 		local player = game.get_player(player_name)
@@ -719,9 +722,12 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t
 			return
 		end
 		-- Two copies can reach here out of order (applyRecipeNotificationDelta awaits zlib). A held finished copy of the
-		-- same generation is replaced: it can be one the controller lost (a crash before its save), and the newer copy
-		-- of the same number is what the controller holds now
-		if (active and active.generation >= generation) or (finished and finished.generation > generation) then
+		-- 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, so this run's copy replaces it whatever the generation
+		local active_newer = active and active.token == token and active.generation >= generation
+		local finished_newer = finished and finished.token == token 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
@@ -732,6 +738,7 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t
 			generation = generation,
 			data = "",
 			prefetch = true,
+			token = token,
 			components = components and compat.json_to_table(components),
 		}
 	end

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() {
@@ -404,7 +407,8 @@ describe("player_sync controller", function() {
 		}
 		const data = generation => ({ name: "test", generation });
 		async function stored(controller) {
-			return (await mock.getHandler(controller, msg.DownloadRequest)(new msg.DownloadRequest(2, "test"))).playerData;
+			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() {
@@ -415,6 +419,24 @@ describe("player_sync controller", function() {
 			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" }));

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, PrefetchEvent } 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";
 
@@ -271,13 +271,13 @@ describe("player_sync", function() {
 			}
 		});
 
-		const prefetch = playerData => mock.getHandler(instance, PrefetchEvent)(new PrefetchEvent(playerData));
+		const prefetch = playerData => mock.getHandler(instance, PrefetchEvent)(new PrefetchEvent(playerData, "tok"));
 		function prefetchCommands(playerData, enabled = Object.keys(components)) {
 			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}'` : ""})`);
+				`${i === 0 ? `,'${sent}','tok'` : ""})`);
 		}
 
 		it("should send prefetch_inventory chunks with the generation and the full recipe list", async function() {
@@ -296,6 +296,12 @@ describe("player_sync", function() {
 			assert.ok(commands.length > 1, "more than one chunk");
 			assert.deepEqual(instance.server.rconCommands, commands);
 		});
+		it("should send the token with the first chunk only", async function() {
+			await prefetch({ generation: 3, name: "test" });
+			const commands = instance.server.rconCommands;
+			assert.ok(commands[0].endsWith(",'tok')"));
+			assert.ok(commands.slice(1).every(command => !command.includes("tok")));
+		});
 		it("should stop when the save refuses the first chunk", async function() {
 			const playerData = { generation: 3, name: "test" };
 			const commands = prefetchCommands(playerData);
@@ -312,4 +318,22 @@ describe("player_sync", function() {
 			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);
+		});
+	});
 });