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) <noreply@anthropic.com>

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);
+			}
 		}
 	});
 

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,

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<string, boolean> | 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<typeof UploadRequest.jsonSchema>) {
-		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<typeof PrefetchEvent.jsonSchema>): PrefetchEvent {
+		return new this(json.playerData as IpcPlayerData);
 	}
 }
 

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));
+		});
+	});
 });