From b8c1e89dff82af64e0c27895c66dfccc8d0407d9 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Sat, 17 Oct 2026 01:42:26 +1100 Subject: [PATCH 01/13] Forward player data to the instance a player is switching to An upload can carry destinationInstanceId. Once the upload is stored, the controller sends the stored data, the upload merged with its copy, in a PrefetchEvent to that instance, if it is another running instance with player_sync enabled. No lock is taken or released. A failed send is logged and doesn't fail the upload. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/controller.ts | 27 +++++++++- plugins/player_sync/index.ts | 1 + plugins/player_sync/messages.ts | 27 +++++++++- plugins/player_sync/test/controller.js | 71 ++++++++++++++++++++++++++ 4 files changed, 124 insertions(+), 2 deletions(-) diff --git a/plugins/player_sync/controller.ts b/plugins/player_sync/controller.ts index 16589fdd..60a91ad1 100644 --- a/plugins/player_sync/controller.ts +++ b/plugins/player_sync/controller.ts @@ -113,6 +113,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)); + 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)) { @@ -225,9 +245,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); + } } }); diff --git a/plugins/player_sync/index.ts b/plugins/player_sync/index.ts index 11d0e0b1..23a2a4f9 100644 --- a/plugins/player_sync/index.ts +++ b/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, diff --git a/plugins/player_sync/messages.ts b/plugins/player_sync/messages.ts index 146d2727..538037bc 100644 --- a/plugins/player_sync/messages.ts +++ b/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 | unknown[], + destination_instance_id?: number, // set_destination's instance, on the leave upload only; not player data } // Marks tombstones in player_sync.json @@ -268,6 +269,7 @@ export class UploadRequest { public instanceId: number, public playerName: string, public playerData: IpcPlayerData, + public destinationInstanceId?: number, ) { } @@ -275,10 +277,33 @@ export class UploadRequest { "instanceId": Type.Integer(), "playerName": Type.String(), "playerData": jsonPlayerData, + "destinationInstanceId": Type.Optional(Type.Integer()), }); static fromJSON(json: Static) { - 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, + ) { + } + + static jsonSchema = Type.Object({ + "playerData": jsonPlayerData, + }); + + static fromJSON(json: Static): PrefetchEvent { + return new this(json.playerData as IpcPlayerData); } } diff --git a/plugins/player_sync/test/controller.js b/plugins/player_sync/test/controller.js index 2a5328df..0f83d51e 100644 --- a/plugins/player_sync/test/controller.js +++ b/plugins/player_sync/test/controller.js @@ -386,4 +386,75 @@ 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) { + return (await mock.getHandler(controller, msg.DownloadRequest)(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 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)); + }); + }); }); From 240180d847ad526fe27bcd782e06080f8e930cee Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Sat, 17 Oct 2026 01:44:19 +1100 Subject: [PATCH 02/13] Stream prefetched player data into the destination save The instance passes destination_instance_id from a leave upload on as destinationInstanceId, and sends a PrefetchEvent's data to player_sync.prefetch_inventory in RCON chunks, using the download's chunk loop. As for a download, the data is stripped by the instance's own player_sync.sync_* options and the components go with the first chunk. A non-empty reply means the save refused or ignored the copy and stops the transfer. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/instance.ts | 74 ++++++++++++++++------ plugins/player_sync/test/instance.js | 94 +++++++++++++++++++++++++++- 2 files changed, 148 insertions(+), 20 deletions(-) diff --git a/plugins/player_sync/instance.ts b/plugins/player_sync/instance.ts index d7fcec40..c133faae 100644 --- a/plugins/player_sync/instance.ts +++ b/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"; @@ -172,11 +173,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 +197,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 +246,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 +277,18 @@ 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 { 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 the delta's add + await sendPlayerData(playerData, undefined, (chunk, number, total, extra) => ( + `/sc player_sync.prefetch_inventory('${name}',${generation},'${chunk}',${number},${total}${extra})` + ), true); + } + instance.handle(PrefetchEvent, handlePrefetch); + /** Tell the scenario which components are synced. */ async function sendComponents() { const json = componentsJson(enabledComponents(instance.config)); diff --git a/plugins/player_sync/test/instance.js b/plugins/player_sync/test/instance.js index 269e87dc..f6828929 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 } from "../dist/node/messages.js"; +import { DownloadResponse, PrefetchEvent } from "../dist/node/messages.js"; import { components } from "../dist/node/components.js"; import { plugin as info } from "../dist/node/index.js"; @@ -211,4 +211,96 @@ 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)); + 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}'` : ""})`); + } + + 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"] }) }; + 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 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)); + }); + }); }); From 18afae46865dc19317ae669a4f4cbf611c543169 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Sat, 17 Oct 2026 01:46:30 +1100 Subject: [PATCH 03/13] Add set_destination remote interface to player_sync A plugin that moves a player calls remote.call("player_sync", "set_destination", player_name, instance_id) before connect_to_server. It only records the destination. Every leave clears it, and a leave that uploads with reason switching_servers adds it to the upload as destination_instance_id. The README describes it for plugin authors. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/README.md | 19 ++++++++++++++++ plugins/player_sync/module/player_sync.lua | 26 ++++++++++++++++++++++ plugins/player_sync/module/serialize.lua | 1 + 3 files changed, 46 insertions(+) diff --git a/plugins/player_sync/README.md b/plugins/player_sync/README.md index 850f1724..5f7c16c3 100644 --- a/plugins/player_sync/README.md +++ b/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 diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index 7e7718d9..bb9248da 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -83,6 +83,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) @@ -486,6 +504,9 @@ 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 @@ -506,6 +527,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, diff --git a/plugins/player_sync/module/serialize.lua b/plugins/player_sync/module/serialize.lua index 8659bb1d..41d37f4f 100644 --- a/plugins/player_sync/module/serialize.lua +++ b/plugins/player_sync/module/serialize.lua @@ -959,6 +959,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 From fe416da26c2221393e04e4c406fb90200975ccaa Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Sat, 17 Oct 2026 01:49:32 +1100 Subject: [PATCH 04/13] Apply a prefetched copy on join without downloading player_sync.prefetch_inventory holds a copy sent ahead in active_downloads and finished_downloads, for a player who is offline or doesn't exist on the destination, with the components its first chunk names. Chunk 1 refuses a player who is online, a download in progress, a stream of the same or a newer generation and a finished copy of a newer one; a finished copy of the same generation is replaced. At the acquire reply a copy is used only when the reply is acquired with its generation and its components are the ones the scenario syncs now. The player is then switched to spectator as a download does and the copy is applied, at once or at its last chunk. Otherwise the copy is dropped and the join continues as before. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../player_sync/module/download_inventory.lua | 32 +++-- plugins/player_sync/module/player_sync.lua | 133 +++++++++++++++--- 2 files changed, 133 insertions(+), 32 deletions(-) diff --git a/plugins/player_sync/module/download_inventory.lua b/plugins/player_sync/module/download_inventory.lua index 2d5a07b4..9e883496 100644 --- a/plugins/player_sync/module/download_inventory.lua +++ b/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 diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index bb9248da..b9a7d790 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/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 @@ -321,6 +351,19 @@ 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 the copy was sent with the components synced here now. Otherwise it goes, and the join carries on + -- as if it had never come: today's download or failure dialog, never the restart path below. + 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 serialize.same_components(record.components, script_data.components) + ) then + downloads[acquire_response.player_name] = nil + end + end + local finished_record = script_data.finished_downloads[acquire_response.player_name] if finished_record then if @@ -328,6 +371,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 @@ -345,6 +396,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 @@ -430,6 +488,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 @@ -451,6 +510,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 @@ -510,6 +570,8 @@ player_sync.events[defines.events.on_pre_player_left_game] = function(event) -- If player has an active download release our acquisition but let the download continue if script_data.active_downloads[player.name] then + -- A prefetched copy is confirmed per join: the next join's acquire reply confirms it again + script_data.active_downloads[player.name].confirmed = nil clusterio_api.send_json("player_sync_release", { player_name = player.name }) return end @@ -589,29 +651,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 @@ -658,4 +698,51 @@ 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 +-- 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) + 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: 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 + rcon.print("Refused prefetch for " .. player_name .. ": generation " .. generation .. " is not newer") + return + end + 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, + components = components and compat.json_to_table(components), + } + end + + local record = script_data.active_downloads[player_name] + if not (record and record.prefetch and record.generation == generation) 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 From ca7caedfbeaeceb9d0c578dad3fcc584021c6bd7 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Sun, 18 Oct 2026 22:09:26 +1100 Subject: [PATCH 05/13] Restore recipe notifications a prefetched copy would leave cleared A prefetched copy is sent without the save's list of cleared recipes, so its delta had only add, and a recipe cleared on the destination but not in the stored list stayed cleared where a download would give it its notification back. Without a current list the instance now sends the whole stored list marked full, and the save gives the recipes it has cleared that aren't in it their notification back, as a download's remove does. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/instance.ts | 13 ++++++++----- plugins/player_sync/module/serialize.lua | 18 ++++++++++++++++++ plugins/player_sync/test/instance.js | 19 ++++++++++++++----- 3 files changed, 40 insertions(+), 10 deletions(-) diff --git a/plugins/player_sync/instance.ts b/plugins/player_sync/instance.ts index c133faae..50543bf7 100644 --- a/plugins/player_sync/instance.ts +++ b/plugins/player_sync/instance.ts @@ -61,6 +61,8 @@ async function encodeLuaString(text: string): Promise { 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, } /** @@ -87,6 +89,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, @@ -97,10 +100,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}`); @@ -282,7 +285,7 @@ export default async function(context: InstancePluginContext) { const { playerData } = 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 the delta's add + // 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); diff --git a/plugins/player_sync/module/serialize.lua b/plugins/player_sync/module/serialize.lua index 41d37f4f..7a79d9ed 100644 --- a/plugins/player_sync/module/serialize.lua +++ b/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 diff --git a/plugins/player_sync/test/instance.js b/plugins/player_sync/test/instance.js index f6828929..a31d5e1e 100644 --- a/plugins/player_sync/test/instance.js +++ b/plugins/player_sync/test/instance.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() { @@ -275,7 +282,9 @@ describe("player_sync", function() { 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"] }) }; + 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)); }); From 5e4dcada508038d9af8240914fb97290337616d2 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Mon, 19 Oct 2026 01:29:19 +1100 Subject: [PATCH 06/13] 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) --- plugins/player_sync/controller.ts | 6 ++++- plugins/player_sync/instance.ts | 12 ++++++--- plugins/player_sync/messages.ts | 9 +++++-- plugins/player_sync/module/player_sync.lua | 19 +++++++++----- plugins/player_sync/test/controller.js | 26 +++++++++++++++++-- plugins/player_sync/test/instance.js | 30 +++++++++++++++++++--- 6 files changed, 84 insertions(+), 18 deletions(-) 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); + }); + }); }); From 17f029e39ead2b8cd3d7bf8bffc2a4ac4ed55be3 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Mon, 19 Oct 2026 01:30:07 +1100 Subject: [PATCH 07/13] Check the run token on every prefetched chunk Chunks of an earlier controller run's copy still in flight could land in a newer run's copy of the same generation. Every chunk now carries the token, and the scenario ignores a chunk whose token differs from its record's. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/instance.ts | 3 ++- plugins/player_sync/module/player_sync.lua | 10 +++++----- plugins/player_sync/test/instance.js | 6 +++--- 3 files changed, 10 insertions(+), 9 deletions(-) diff --git a/plugins/player_sync/instance.ts b/plugins/player_sync/instance.ts index 7a64164b..8241d1ec 100644 --- a/plugins/player_sync/instance.ts +++ b/plugins/player_sync/instance.ts @@ -289,7 +289,8 @@ export default async function(context: InstancePluginContext) { 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) => { - const tokenArg = number === 1 ? `,'${token}'` : ""; + // Later chunks have no components argument, so nil holds its place + const tokenArg = number === 1 ? `,'${token}'` : `,nil,'${token}'`; const args = `'${name}',${generation},'${chunk}',${number},${total}${extra}${tokenArg}`; return `/sc player_sync.prefetch_inventory(${args})`; }, true); diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index d344db8e..3459e89f 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -701,11 +701,11 @@ end -- 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. +-- server. It holds the copy for a player who is offline here or doesn't exist yet. Every chunk carries the token, so +-- chunks of an earlier controller run's copy never land in a newer 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, given with the first chunk +--- @param token string? The controller run that sent the copy function player_sync.prefetch_inventory(player_name, generation, data, number, total, components, token) local script_data = get_script_data() if number == 1 then @@ -744,7 +744,7 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t end local record = script_data.active_downloads[player_name] - if not (record and record.prefetch and record.generation == generation) then + if not (record and record.prefetch and record.generation == generation and record.token == token) 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 diff --git a/plugins/player_sync/test/instance.js b/plugins/player_sync/test/instance.js index 7a7778c6..b9568af7 100644 --- a/plugins/player_sync/test/instance.js +++ b/plugins/player_sync/test/instance.js @@ -277,7 +277,7 @@ describe("player_sync", function() { 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'` : ""})`); + `${i === 0 ? `,'${sent}','tok'` : ",nil,'tok'"})`); } it("should send prefetch_inventory chunks with the generation and the full recipe list", async function() { @@ -296,11 +296,11 @@ 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() { + it("should send the token with every chunk", 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"))); + assert.ok(commands.length > 1 && commands.slice(1).every(command => command.endsWith(",nil,'tok')"))); }); it("should stop when the save refuses the first chunk", async function() { const playerData = { generation: 3, name: "test" }; From fdec6bc249eff64909d2889d85cba0773126982a Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Mon, 19 Oct 2026 01:57:27 +1100 Subject: [PATCH 08/13] Explain why a prefetched copy needs the run token as well as the generation Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/module/player_sync.lua | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index 3459e89f..cfbe533e 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -352,8 +352,11 @@ function player_sync.sync_player(acquire_response) -- 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 the copy was sent with the components synced here now. Otherwise it goes, and the join carries on - -- as if it had never come: today's download or failure dialog, never the restart path below. + -- generation, 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. 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 ( From 0bd7fccdf37b0c9b0554afef60c67cf72c85ef2f Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Mon, 19 Oct 2026 08:58:01 +1100 Subject: [PATCH 09/13] Keep newer local player data over a prefetched copy An instance with older data can upload the generation a save already holds, and a prefetch of it then overwrote that save's newer state. A join now drops a prefetched copy unless the ordinary download check says the data is needed. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/module/player_sync.lua | 30 +++++++++++++--------- 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index cfbe533e..7d5cfcbb 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -334,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)) @@ -356,14 +369,15 @@ function player_sync.sync_player(acquire_response) -- 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. + -- path below, which only sees a different generation. 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. 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 ( + 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 + ) or not download_needed(script_data.players[acquire_response.player_name], acquire_response, script_data)) then downloads[acquire_response.player_name] = nil end end @@ -458,15 +472,7 @@ 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 download_needed(player_record, acquire_response, script_data) then player_sync.initiate_inventory_download(player, player_record, acquire_response.generation, epoch) else From 9c4e35473811a5ab8cd78fb05bb1d357b2f8ccc5 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Tue, 20 Oct 2026 12:27:34 +1100 Subject: [PATCH 10/13] Check the epoch on prefetched copies An admin import or delete bumps the stored data's epoch and can start the generations over, so a copy held from before it must not apply. The prefetch carries the epoch like the token, and a held copy is used only when it matches the acquire reply's. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/instance.ts | 5 ++- plugins/player_sync/module/player_sync.lua | 46 +++++++++++++--------- plugins/player_sync/test/instance.js | 15 ++++--- 3 files changed, 41 insertions(+), 25 deletions(-) diff --git a/plugins/player_sync/instance.ts b/plugins/player_sync/instance.ts index 8241d1ec..83895621 100644 --- a/plugins/player_sync/instance.ts +++ b/plugins/player_sync/instance.ts @@ -286,12 +286,13 @@ export default async function(context: InstancePluginContext) { 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 tokenArg = number === 1 ? `,'${token}'` : `,nil,'${token}'`; - const args = `'${name}',${generation},'${chunk}',${number},${total}${extra}${tokenArg}`; + 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); } diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index 7d5cfcbb..35ffbe85 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -334,8 +334,8 @@ 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. +-- 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 ( @@ -365,17 +365,17 @@ function player_sync.sync_player(acquire_response) -- 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, 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. 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. + -- 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. 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. 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.token == acquire_response.token + 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 downloads[acquire_response.player_name] = nil @@ -709,13 +709,15 @@ function player_sync.check_active_uploads() end -- 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. Every chunk carries the token, so --- chunks of an earlier controller run's copy never land in a newer one. It prints only when it refuses or ignores a --- chunk, which stops the host sending the rest. +-- 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 -function player_sync.prefetch_inventory(player_name, generation, data, number, total, components, token) +--- @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) @@ -733,9 +735,13 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t -- 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, 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 + -- 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 @@ -748,12 +754,16 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t data = "", prefetch = true, token = token, + epoch = epoch, components = components and compat.json_to_table(components), } end local record = script_data.active_downloads[player_name] - if not (record and record.prefetch and record.generation == generation and record.token == token) then + 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 diff --git a/plugins/player_sync/test/instance.js b/plugins/player_sync/test/instance.js index b9568af7..12209271 100644 --- a/plugins/player_sync/test/instance.js +++ b/plugins/player_sync/test/instance.js @@ -272,12 +272,12 @@ describe("player_sync", function() { }); const prefetch = playerData => mock.getHandler(instance, PrefetchEvent)(new PrefetchEvent(playerData, "tok")); - function prefetchCommands(playerData, enabled = Object.keys(components)) { + 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'` : ",nil,'tok'"})`); + `${i === 0 ? `,'${sent}','tok',${epoch}` : `,nil,'tok',${epoch}`})`); } it("should send prefetch_inventory chunks with the generation and the full recipe list", async function() { @@ -296,11 +296,16 @@ 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 every chunk", async function() { + 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')")); - assert.ok(commands.length > 1 && commands.slice(1).every(command => command.endsWith(",nil,'tok')"))); + 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" }; From 3f79cdb1a65adbf41711683da784465753fb24e8 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Tue, 20 Oct 2026 12:28:08 +1100 Subject: [PATCH 11/13] Download when a dropped prefetched copy already took the body A copy confirmed at an earlier join has replaced the body with a spectator. If a later join drops it (another controller run) while the generations match, the join kept the local state: a spectator with no body, uploaded on the next leave. Download instead. A newer copy replacing a held one keeps where the body stood, so the replacement still counts as having taken it. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/module/player_sync.lua | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index 35ffbe85..4a02932c 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -369,8 +369,12 @@ function player_sync.sync_player(acquire_response) -- 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. 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. + -- 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 = false 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 ( @@ -378,6 +382,7 @@ function player_sync.sync_player(acquire_response) 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 + body_taken = body_taken or record.position ~= nil downloads[acquire_response.player_name] = nil end end @@ -472,7 +477,7 @@ function player_sync.sync_player(acquire_response) end local stale = (player_record.epoch or 0) ~= epoch - if download_needed(player_record, acquire_response, script_data) 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) else @@ -746,6 +751,9 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t 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, @@ -756,6 +764,9 @@ function player_sync.prefetch_inventory(player_name, generation, data, number, t 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 From 0149c72639bcf6c6cf324c74a512b0f1989b8be1 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Tue, 20 Oct 2026 13:19:29 +1100 Subject: [PATCH 12/13] Upload on leave while an unconfirmed prefetched copy is held The leave handler skipped the upload for any active download, including a prefetched copy no join had confirmed, which has replaced nothing here. Newer local data, such as after a failed upload whose retry the join cleared, was then never uploaded. Only a real download skips it now; a copy whose join took the body still uploads nothing through sync_in_progress. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/module/player_sync.lua | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index 4a02932c..a54d2633 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -583,10 +583,15 @@ player_sync.events[defines.events.on_pre_player_left_game] = function(event) 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 - -- A prefetched copy is confirmed per join: the next join's acquire reply confirms it again - script_data.active_downloads[player.name].confirmed = nil + 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 From cb450690ba3bbcfa25f2480ad922446b255b6122 Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Tue, 20 Oct 2026 13:19:49 +1100 Subject: [PATCH 13/13] Restore the body's place on the download forced by a dropped copy The download a join forces after dropping a copy that took the body found the player already a spectator, so it kept the spectator's place and restored the character there. It now takes the surface, position and vehicle the dropped copy kept. Co-Authored-By: Claude Opus 5.5 (1M context) --- plugins/player_sync/module/player_sync.lua | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/plugins/player_sync/module/player_sync.lua b/plugins/player_sync/module/player_sync.lua index a54d2633..1ee077f2 100644 --- a/plugins/player_sync/module/player_sync.lua +++ b/plugins/player_sync/module/player_sync.lua @@ -374,7 +374,7 @@ function player_sync.sync_player(acquire_response) -- 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 = false + 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 ( @@ -382,7 +382,9 @@ function player_sync.sync_player(acquire_response) 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 - body_taken = body_taken or record.position ~= nil + if record.position ~= nil then + body_taken = record + end downloads[acquire_response.player_name] = nil end end @@ -479,6 +481,13 @@ function player_sync.sync_player(acquire_response) local stale = (player_record.epoch or 0) ~= epoch 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