From 240180d847ad526fe27bcd782e06080f8e930cee Mon Sep 17 00:00:00 2001 From: ILLISIS Date: Sat, 17 Oct 2026 01:44:19 +1100 Subject: [PATCH] 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)); + }); + }); });