diff --git a/plugins/player_sync/controller.ts b/plugins/player_sync/controller.ts index 60a91ad1..513e3a2b 100644 --- a/plugins/player_sync/controller.ts +++ b/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(); 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, ); }); diff --git a/plugins/player_sync/instance.ts b/plugins/player_sync/instance.ts index 50543bf7..7a64164b 100644 --- a/plugins/player_sync/instance.ts +++ b/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); diff --git a/plugins/player_sync/messages.ts b/plugins/player_sync/messages.ts index 538037bc..0f12a9ce 100644 --- a/plugins/player_sync/messages.ts +++ b/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): 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): PrefetchEvent { - return new this(json.playerData as IpcPlayerData); + return new this(json.playerData as IpcPlayerData, json.token); } } diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index b9a7d790..d344db8e 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/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 diff --git a/plugins/player_sync/test/controller.js b/plugins/player_sync/test/controller.js index 0f83d51e..76eb0f27 100644 --- a/plugins/player_sync/test/controller.js +++ b/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" })); diff --git a/plugins/player_sync/test/instance.js b/plugins/player_sync/test/instance.js index a31d5e1e..7a7778c6 100644 --- a/plugins/player_sync/test/instance.js +++ b/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); + }); + }); });