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/controller.ts b/plugins/player_sync/controller.ts index 16589fdd..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; @@ -113,6 +116,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, 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 + 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)) { @@ -166,6 +189,7 @@ export default async function(context: ControllerPluginContext) { playerDatastore.has(playerName), undefined, epoch || undefined, + runToken, ); }); @@ -225,9 +249,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/instance.ts b/plugins/player_sync/instance.ts index d7fcec40..83895621 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"; @@ -28,6 +29,7 @@ type IpcAcquireResponse = { has_data?: boolean, message?: string, epoch?: number, + token?: string, } /** @@ -60,6 +62,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, } /** @@ -86,6 +90,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, @@ -96,10 +101,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}`); @@ -135,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)) { @@ -172,11 +178,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 +202,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 +251,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 +282,22 @@ 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, 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 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); + } + 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/messages.ts b/plugins/player_sync/messages.ts index 146d2727..0f12a9ce 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 @@ -195,6 +196,7 @@ export class AcquireResponse { public hasData?: boolean, public message?: string, public epoch?: number, + public token?: string, ) { } @@ -204,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 { @@ -268,6 +271,7 @@ export class UploadRequest { public instanceId: number, public playerName: string, public playerData: IpcPlayerData, + public destinationInstanceId?: number, ) { } @@ -275,10 +279,36 @@ 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, + /** 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, json.token); } } 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 7e7718d9..1ee077f2 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 @@ -83,6 +113,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) @@ -286,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)) @@ -303,6 +364,31 @@ 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 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 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 -- 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 ( + acquire_response.status == "acquired" and record.generation == acquire_response.generation + 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 + if record.position ~= nil then + body_taken = record + end + downloads[acquire_response.player_name] = nil + end + end + local finished_record = script_data.finished_downloads[acquire_response.player_name] if finished_record then if @@ -310,6 +396,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 @@ -327,6 +421,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 @@ -378,16 +479,15 @@ 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 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 @@ -412,6 +512,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 @@ -433,6 +534,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 @@ -486,9 +588,19 @@ 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 + 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 @@ -506,6 +618,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, @@ -563,29 +680,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 @@ -632,4 +727,73 @@ function player_sync.check_active_uploads() end end +-- Called by instance.ts as /sc player_sync.prefetch_inventory('name', generation, 'chunk', number, total, components, +-- 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 +--- @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) + 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: 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, 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 + 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, + last_active = game.ticks_played, + generation = generation, + data = "", + prefetch = true, + 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 + + local record = script_data.active_downloads[player_name] + 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 + end + download_inventory(player_name, data, number, total) +end + return player_sync diff --git a/plugins/player_sync/module/serialize.lua b/plugins/player_sync/module/serialize.lua index 8659bb1d..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 @@ -959,6 +977,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 diff --git a/plugins/player_sync/test/controller.js b/plugins/player_sync/test/controller.js index 2a5328df..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() { @@ -386,4 +389,94 @@ 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) { + 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() { + 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 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" })); + 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)); + }); + }); }); diff --git a/plugins/player_sync/test/instance.js b/plugins/player_sync/test/instance.js index 269e87dc..12209271 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 { AcquireResponse, DownloadResponse, PrefetchEvent } from "../dist/node/messages.js"; import { components } from "../dist/node/components.js"; import { plugin as info } from "../dist/node/index.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() { @@ -211,4 +218,127 @@ 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, "tok")); + 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',${epoch}` : `,nil,'tok',${epoch}`})`); + } + + 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"], full: true }), + }; + 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 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',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" }; + 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)); + }); + }); + + 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); + }); + }); });