From f12b016cec4d9fc8c2b61c2e784e11d8f040b0dc Mon Sep 17 00:00:00 2001 From: bbassie <17990055+bbassie@users.noreply.github.com> Date: Sat, 17 Oct 2026 15:32:20 +0000 Subject: [PATCH] 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 --- plugins/player_sync/README.md | 7 +- plugins/player_sync/controller.ts | 75 ++++++++++++------- plugins/player_sync/instance.ts | 3 + plugins/player_sync/messages.ts | 9 ++- .../player_sync/module/download_inventory.lua | 3 +- plugins/player_sync/module/player_sync.lua | 28 ++++++- plugins/player_sync/module/serialize.lua | 1 + plugins/player_sync/test/controller.js | 55 +++++++++++++- plugins/player_sync/test/plugin.js | 17 +++++ 9 files changed, 159 insertions(+), 39 deletions(-) diff --git a/plugins/player_sync/README.md b/plugins/player_sync/README.md index 0c4a1a9c..e3e28153 100644 --- a/plugins/player_sync/README.md +++ b/plugins/player_sync/README.md @@ -45,8 +45,11 @@ The same operations are available from clusterioctl: npx clusterioctl player-sync delete npx clusterioctl player-sync release -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. diff --git a/plugins/player_sync/controller.ts b/plugins/player_sync/controller.ts index 1355a18f..902dc179 100644 --- a/plugins/player_sync/controller.ts +++ b/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(itemsPath, logger); +} + +async function loadMap(file: string, logger: lib.Logger): Promise> { + 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 | 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, 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(); + const databaseDirectory = controller.config.get("controller.database_directory"); const playerDatastore: Map = await loadDatabase(controller.config, logger); + const tombstonesPath = path.resolve(databaseDirectory, "player_sync_deleted.json"); + const tombstones = await loadMap(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}`); diff --git a/plugins/player_sync/instance.ts b/plugins/player_sync/instance.ts index b4e6b978..d7fcec40 100644 --- a/plugins/player_sync/instance.ts +++ b/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)) { diff --git a/plugins/player_sync/messages.ts b/plugins/player_sync/messages.ts index 6a5433b2..07019686 100644 --- a/plugins/player_sync/messages.ts +++ b/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): 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 { diff --git a/plugins/player_sync/module/download_inventory.lua b/plugins/player_sync/module/download_inventory.lua index d6f6e54d..2d5a07b4 100644 --- a/plugins/player_sync/module/download_inventory.lua +++ b/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 diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index 6285c8fe..1a01600d 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/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() diff --git a/plugins/player_sync/module/serialize.lua b/plugins/player_sync/module/serialize.lua index 7f9b3b2a..a2176ebf 100644 --- a/plugins/player_sync/module/serialize.lua +++ b/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? diff --git a/plugins/player_sync/test/controller.js b/plugins/player_sync/test/controller.js index 67b9c4c7..b555031d 100644 --- a/plugins/player_sync/test/controller.js +++ b/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")); }); diff --git a/plugins/player_sync/test/plugin.js b/plugins/player_sync/test/plugin.js index 3c0e5f7a..3c30f6be 100644 --- a/plugins/player_sync/test/plugin.js +++ b/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])); });