Invalidate instance copies when player data is imported or deleted

Generations alone could not order a repair against copies held by
instances. Deleting dropped the watermark, so a stale instance could
skip downloading the new baseline. An import could land on the same
generation as an upload that never reached the controller, and the
instance would then keep its own copy.

Imports and deletes now bump a per-player epoch, which the controller
returns on acquire and checks on upload. Instances throw away a copy
from an earlier epoch, and deleted players keep their generation and
epoch in player_sync_deleted.json. A restarted download also takes the
new generation instead of the outdated one.

Generations in player data must now be non-negative integers, so an
invalid import is refused before it reaches the datastore.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/player_sync/README.md

@@ -45,8 +45,11 @@ The same operations are available from clusterioctl:
     npx clusterioctl player-sync delete <player>
     npx clusterioctl player-sync release <player>
 
-Importing bumps the generation so instances holding an older copy download the new one the next time the player joins.
-Deleting makes the next join turn whatever the player has on that instance into the new synced inventory.
+Importing and deleting both start a new epoch for the player.
+An instance holding a copy from an earlier epoch throws it away the next time the player joins, even if it is newer than what the controller has, and an upload from an earlier epoch is discarded.
+After an import the instance downloads the new data.
+After a delete the next join turns whatever the player has on that instance into the new synced inventory.
+Deleted players stay listed in `player_sync_deleted.json` in the database directory so their epoch survives a controller restart.
 Both are refused while an instance holds the lock on the player, because the instance uploads its own copy when the player leaves and that would undo the change.
 Have the player leave first, or release the lock if the instance is stuck holding it.
 

plugins/player_sync/controller.ts

@@ -22,39 +22,47 @@ async function loadDatabase(config: lib.ControllerConfig, logger: lib.Logger) {
 			}
 		}
 	}
-	logger.verbose(`Loading ${itemsPath}`);
+	return await loadMap<IpcPlayerData>(itemsPath, logger);
+}
+
+async function loadMap<V>(file: string, logger: lib.Logger): Promise<Map<string, V>> {
+	logger.verbose(`Loading ${file}`);
 	try {
-		let content = await fs.readFile(itemsPath, { encoding: "utf8" });
+		let content = await fs.readFile(file, { encoding: "utf8" });
 		return new Map(JSON.parse(content));
 
 	} catch (err: any) {
 		if (err.code === "ENOENT") {
-			logger.verbose("Creating new player data database");
+			logger.verbose(`Creating new ${path.basename(file)}`);
 			return new Map();
 		}
 		throw err;
 	}
 }
 
-async function saveDatabase(
-	controllerConfig: lib.ControllerConfig,
-	playerDatastore: Map<string, IpcPlayerData> | undefined,
-	logger: lib.Logger,
-) {
-	if (playerDatastore) {
-		let file = path.resolve(controllerConfig.get("controller.database_directory"), "player_sync.json");
-		logger.verbose(`writing ${file}`);
-		let content = JSON.stringify(Array.from(playerDatastore));
-		await lib.safeOutputFile(file, content);
-	}
+async function saveMap(file: string, map: Map<string, unknown>, logger: lib.Logger) {
+	logger.verbose(`writing ${file}`);
+	await lib.safeOutputFile(file, JSON.stringify(Array.from(map)));
 }
 
+// Left behind by deleted player data so instances can tell their copy is stale
+type Tombstone = { generation: number, epoch: number };
+
 export default async function(context: ControllerPluginContext) {
 	const { controller, logger, plugin } = context;
 	const acquiredPlayers = new Map<string, { instanceId: number, expiresMs?: number }>();
+	const databaseDirectory = controller.config.get("controller.database_directory");
 	const playerDatastore: Map<string, IpcPlayerData> = await loadDatabase(controller.config, logger);
+	const tombstonesPath = path.resolve(databaseDirectory, "player_sync_deleted.json");
+	const tombstones = await loadMap<Tombstone>(tombstonesPath, logger);
 	let playerDatastoreDirty = false;
 
+	// Generation and epoch instances should compare their copy against
+	function storedVersion(playerName: string) {
+		let stored = playerDatastore.get(playerName) ?? tombstones.get(playerName);
+		return { generation: stored?.generation ?? 0, epoch: stored?.epoch ?? 0 };
+	}
+
 	// Returns the acquisition record for the player if it's still in effect
 	function getAcquisition(playerName: string) {
 		let acquisitionRecord = acquiredPlayers.get(playerName);
@@ -122,7 +130,8 @@ export default async function(context: ControllerPluginContext) {
 	controller.hooks.save.attach(plugin.name, async () => {
 		if (playerDatastoreDirty) {
 			playerDatastoreDirty = false;
-			await saveDatabase(controller.config, playerDatastore, logger);
+			await saveMap(path.resolve(databaseDirectory, "player_sync.json"), playerDatastore, logger);
+			await saveMap(tombstonesPath, tombstones, logger);
 		}
 	});
 
@@ -137,11 +146,13 @@ export default async function(context: ControllerPluginContext) {
 			};
 		}
 
-		let playerData = playerDatastore.get(playerName);
+		let { generation, epoch } = storedVersion(playerName);
 		return new msg.AcquireRequest.Response(
 			"acquired",
-			playerData ? playerData.generation : 0,
-			Boolean(playerData),
+			generation,
+			playerDatastore.has(playerName),
+			undefined,
+			epoch || undefined,
 		);
 	});
 
@@ -176,18 +187,26 @@ export default async function(context: ControllerPluginContext) {
 		}
 
 		acquiredPlayers.delete(playerName);
-		let oldPlayerData = playerDatastore.get(playerName);
-		if (store && oldPlayerData && oldPlayerData.generation >= playerData.generation) {
+		let stored = storedVersion(playerName);
+		if (store && (playerData.epoch ?? 0) !== stored.epoch) {
 			logger.warn(
-				`${instanceName} uploaded generation ${playerData.generation} while the stored` +
-				`generation is ${oldPlayerData.generation} for ${playerName}`
+				`${instanceName} uploaded ${playerName} from epoch ${playerData.epoch ?? 0} while the stored ` +
+				`epoch is ${stored.epoch}, discarding the outdated copy`
+			);
+			store = false;
+
+		} else if (store && stored.generation >= playerData.generation) {
+			logger.warn(
+				`${instanceName} uploaded generation ${playerData.generation} while the stored ` +
+				`generation is ${stored.generation} for ${playerName}`
 			);
 			store = false;
 		}
 
 		if (store) {
 			logger.verbose(`Received player data for ${playerName} from ${instanceName}`);
-			playerDatastore.set(playerName, mergePlayerData(oldPlayerData, playerData));
+			playerDatastore.set(playerName, mergePlayerData(playerDatastore.get(playerName), playerData));
+			tombstones.delete(playerName);
 			playerDatastoreDirty = true;
 		}
 	});
@@ -253,11 +272,13 @@ export default async function(context: ControllerPluginContext) {
 		let { playerName, playerData } = request;
 		checkNotAcquired(playerName);
 
-		// Bump the generation so instances holding an older copy download this one
-		let oldPlayerData = playerDatastore.get(playerName);
+		// A new epoch makes every instance drop its copy, including ones with an unconfirmed upload
+		let stored = storedVersion(playerName);
 		playerData.name = playerName;
-		playerData.generation = Math.max(playerData.generation, (oldPlayerData?.generation ?? 0) + 1);
+		playerData.generation = Math.max(playerData.generation, stored.generation + 1);
+		playerData.epoch = stored.epoch + 1;
 		playerDatastore.set(playerName, playerData);
+		tombstones.delete(playerName);
 		playerDatastoreDirty = true;
 		logger.info(`Replaced player data for ${playerName} with generation ${playerData.generation}`);
 		return new msg.SetPlayerDataRequest.Response(playerData.generation);
@@ -270,6 +291,8 @@ export default async function(context: ControllerPluginContext) {
 		}
 		checkNotAcquired(playerName);
 
+		let { generation, epoch } = storedVersion(playerName);
+		tombstones.set(playerName, { generation, epoch: epoch + 1 });
 		playerDatastore.delete(playerName);
 		playerDatastoreDirty = true;
 		logger.info(`Deleted player data for ${playerName}`);

plugins/player_sync/instance.ts

@@ -27,6 +27,7 @@ type IpcAcquireResponse = {
 	generation?: number,
 	has_data?: boolean,
 	message?: string,
+	epoch?: number,
 }
 
 /**
@@ -118,6 +119,7 @@ export default async function(context: InstancePluginContext) {
 			message: "Controller is temporarily unavailable",
 			has_data: undefined,
 			generation: undefined,
+			epoch: undefined,
 		};
 
 		if (host.connector.connected && !disconnecting) {
@@ -132,6 +134,7 @@ export default async function(context: InstancePluginContext) {
 					generation: acquireResponse.generation,
 					has_data: acquireResponse.hasData,
 					message: acquireResponse.message,
+					epoch: acquireResponse.epoch,
 				};
 			} catch (err: any) {
 				if (!(err instanceof lib.SessionLost)) {

plugins/player_sync/messages.ts

@@ -4,6 +4,8 @@ import { StringEnum, jsonArray } from "@clusterio/lib";
 // .\module\serialize.lua:serialize.serialize_player()
 export type IpcPlayerData = {
 	generation: number,
+	/** Bumped when the stored data is imported or deleted, invalidating copies held by instances */
+	epoch?: number,
 	name: string,
 	controller?: string,
 	color?: number[],
@@ -27,7 +29,8 @@ export type IpcPlayerData = {
 }
 
 const jsonPlayerData = Type.Object({
-	generation: Type.Number(),
+	generation: Type.Integer({ minimum: 0 }),
+	epoch: Type.Optional(Type.Integer({ minimum: 0 })),
 	name: Type.String(),
 });
 
@@ -37,6 +40,7 @@ export class AcquireResponse {
 		public generation?: number,
 		public hasData?: boolean,
 		public message?: string,
+		public epoch?: number,
 	) {
 	}
 
@@ -45,10 +49,11 @@ export class AcquireResponse {
 		"generation": Type.Optional(Type.Integer()),
 		"hasData": Type.Optional(Type.Boolean()),
 		"message": Type.Optional(Type.String()),
+		"epoch": Type.Optional(Type.Integer()),
 	});
 
 	static fromJSON(json: Static<typeof AcquireResponse.jsonSchema>): AcquireResponse {
-		return new this(json.status, json.generation, json.hasData, json.message);
+		return new this(json.status, json.generation, json.hasData, json.message, json.epoch);
 	}
 };
 export class AcquireRequest {

plugins/player_sync/module/download_inventory.lua

@@ -29,7 +29,7 @@ local function download_inventory(player_name, data, number, total, components)
 	end
 	if record.restart then
 		rcon.print("Restarting outdated download")
-		player_sync.initiate_inventory_download(player, player_record, record.generation)
+		player_sync.initiate_inventory_download(player, player_record, record.generation, record.epoch)
 		return
 	end
 
@@ -55,6 +55,7 @@ local function download_inventory(player_name, data, number, total, components)
 		player_record.sync = true
 		player_record.components = record.components
 		player_record.generation = record.generation
+		player_record.epoch = record.epoch
 		return
 	end
 

plugins/player_sync/module/player_sync.lua

@@ -99,6 +99,7 @@ function player_sync.serialize_player(player, player_record)
 	local failed_deserialization = get_script_data().failed_deserialization[player.name] or {}
 	local serialized_player = serialize.serialize_player(player, failed_deserialization, player_record.components)
 	serialized_player.generation = player_record.generation
+	serialized_player.epoch = player_record.epoch
 
 	return serialized_player
 end
@@ -269,9 +270,15 @@ function player_sync.sync_player(acquire_response)
 	end
 
 	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
 	local finished_record = script_data.finished_downloads[acquire_response.player_name]
 	if finished_record then
-		if acquire_response.status == "acquired" and finished_record.generation == acquire_response.generation then
+		if
+			acquire_response.status == "acquired"
+			and finished_record.generation == acquire_response.generation
+			and (finished_record.epoch or 0) == epoch
+		then
 			player_sync.finish_download(player, finished_record)
 			return
 		end
@@ -284,7 +291,11 @@ function player_sync.sync_player(acquire_response)
 	local download_record = script_data.active_downloads[acquire_response.player_name]
 	if download_record then
 		-- Ignore the response if it affirms the current active download
-		if acquire_response.status == "acquired" and download_record.generation == acquire_response.generation then
+		if
+			acquire_response.status == "acquired"
+			and download_record.generation == acquire_response.generation
+			and (download_record.epoch or 0) == epoch
+		then
 			game.print("Continuing active download")
 			return
 		end
@@ -305,6 +316,8 @@ function player_sync.sync_player(acquire_response)
 			-- download was started. Restart it.
 			game.print("Restarting download")
 			download_record.restart = true
+			download_record.generation = acquire_response.generation
+			download_record.epoch = epoch
 			return
 		end
 
@@ -333,15 +346,17 @@ function player_sync.sync_player(acquire_response)
 		return
 	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
-		player_sync.initiate_inventory_download(player, player_record, acquire_response.generation)
+		player_sync.initiate_inventory_download(player, player_record, acquire_response.generation, epoch)
 
 	else
 		if acquire_response.generation == 0 then
@@ -355,6 +370,9 @@ function player_sync.sync_player(acquire_response)
 		start_sync(player_record, script_data.components)
 		player_record.dirty = true
 		player_record.sync_in_progress = false
+		-- Deleted data keeps its generation and epoch so the next upload is accepted as the new baseline
+		player_record.epoch = epoch
+		player_record.generation = math.max(player_record.generation, acquire_response.generation)
 	end
 end
 
@@ -378,6 +396,7 @@ function player_sync.finish_download(player, finished_record)
 	player_record.sync_in_progress = false
 	start_sync(player_record, finished_record.components)
 	player_record.generation = finished_record.generation
+	player_record.epoch = finished_record.epoch
 end
 
 -- Download inventory from controller
@@ -478,12 +497,13 @@ function player_sync.confirm_upload(player_name, generation)
 	player_record.dirty = false
 end
 
-function player_sync.initiate_inventory_download(player, player_record, generation)
+function player_sync.initiate_inventory_download(player, player_record, generation, epoch)
 	player.print("Initiating inventory download...")
 	local record = {
 		started = game.ticks_played,
 		last_active = game.ticks_played,
 		generation = generation,
+		epoch = epoch,
 		data = "",
 	}
 	local script_data = get_script_data()

plugins/player_sync/module/serialize.lua

@@ -897,6 +897,7 @@ end
 
 --- @class SerializedPlayerData
 --- @field generation number
+--- @field epoch number?
 --- @field name string
 --- @field controller string?
 --- @field color Color?

plugins/player_sync/test/controller.js

@@ -104,9 +104,15 @@ describe("player_sync controller", function() {
 			const response = await handler(msg.AcquireRequest)(new msg.AcquireRequest(instanceId, playerName));
 			return response.status === "acquired";
 		}
-		async function saved() {
+		async function saved(file = "player_sync.json") {
 			await controller.hooks.save.invoke();
-			return new Map(JSON.parse(await fs.readFile(path.join(databaseDir, "player_sync.json"), "utf8")));
+			return new Map(JSON.parse(await fs.readFile(path.join(databaseDir, file), "utf8")));
+		}
+		async function upload(instanceId, data) {
+			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));
 		}
 
 		beforeEach(async function() {
@@ -170,10 +176,25 @@ describe("player_sync controller", function() {
 			async function release(name) {
 				await handler(msg.ForceReleaseRequest)(new msg.ForceReleaseRequest(name));
 			}
-			it("should store data with a bumped generation", async function() {
+			it("should store data with a bumped generation and epoch", async function() {
 				await release("stored");
 				assert.deepEqual(await set("stored", playerData("other", 1)), new msg.SetPlayerDataResponse(4));
-				assert.deepEqual((await saved()).get("stored"), playerData("stored", 4));
+				assert.deepEqual((await saved()).get("stored"), playerData("stored", 4, { epoch: 1 }));
+			});
+			it("should invalidate unconfirmed uploads from instances", async function() {
+				// Instance 1 left with generation 4 but the upload never reached the controller
+				await release("stored");
+				await set("stored", playerData("stored", 0));
+				await upload(1, playerData("stored", 4));
+				assert.deepEqual((await saved()).get("stored"), playerData("stored", 4, { epoch: 1 }));
+				assert.deepEqual(
+					await acquireResponse(1, "stored"), new msg.AcquireResponse("acquired", 4, true, undefined, 1),
+				);
+			});
+			it("should ignore the epoch in imported data", async function() {
+				await release("stored");
+				await set("stored", playerData("stored", 0, { epoch: 7 }));
+				assert.equal((await saved()).get("stored").epoch, 1);
 			});
 			it("should keep a higher provided generation", async function() {
 				await release("stored");
@@ -198,6 +219,32 @@ describe("player_sync controller", function() {
 				await del("stored");
 				assert(!(await saved()).has("stored"));
 			});
+			it("should keep the generation and bump the epoch", async function() {
+				await handler(msg.ForceReleaseRequest)(new msg.ForceReleaseRequest("stored"));
+				await del("stored");
+				assert.deepEqual(
+					await acquireResponse(2, "stored"), new msg.AcquireResponse("acquired", 3, false, undefined, 1),
+				);
+				assert.deepEqual((await saved("player_sync_deleted.json")).get("stored"), { generation: 3, epoch: 1 });
+
+				controller = await load();
+				assert.deepEqual(
+					await acquireResponse(2, "stored"), new msg.AcquireResponse("acquired", 3, false, undefined, 1),
+				);
+			});
+			it("should accept the next upload from the new epoch only", async function() {
+				await handler(msg.ForceReleaseRequest)(new msg.ForceReleaseRequest("stored"));
+				await del("stored");
+				await upload(1, playerData("stored", 11));
+				assert(!(await saved()).has("stored"));
+				await upload(2, playerData("stored", 4, { epoch: 1 }));
+				assert.deepEqual((await saved()).get("stored"), playerData("stored", 4, { epoch: 1 }));
+				assert(!(await saved("player_sync_deleted.json")).has("stored"));
+
+				// The other instance downloads the baseline instead of uploading its stale copy
+				await upload(1, playerData("stored", 11));
+				assert.equal((await saved()).get("stored").generation, 4);
+			});
 			it("should reject unknown players", async function() {
 				await assert.rejects(del("unknown"), new lib.RequestError("No player data stored for unknown"));
 			});

plugins/player_sync/test/plugin.js

@@ -34,6 +34,23 @@ describe("player_sync plugin", function() {
 				[undefined, 1], // instanceId
 			));
 		});
+		it("AcquireResponse should be round trip json serialisable", function() {
+			testRoundTripJsonSerialisable(msg.AcquireResponse, testMatrix(
+				["acquired", "busy"], // status
+				[undefined, 3], // generation
+				[undefined, true], // hasData
+				[undefined, "Instance"], // message
+				[undefined, 2], // epoch
+			));
+		});
+		it("should reject non-integer generations", function() {
+			const validate = lib.compile(msg.SetPlayerDataRequest.jsonSchema);
+			assert(validate({ playerName: "player", playerData: playerData("player", 5) }));
+			for (const generation of [5.5, -1, "5"]) {
+				const request = { playerName: "player", playerData: playerData("player", generation) };
+				assert(!validate(request), `accepted ${generation}`);
+			}
+		});
 		it("SetPlayerDataResponse should be round trip json serialisable", function() {
 			testRoundTripJsonSerialisable(msg.SetPlayerDataResponse, testMatrix([1, 5]));
 		});