Send only recipe notification changes on inventory download (#997)

* Send only recipe notification changes on inventory download

Lua includes its current cleared recipe list in the download request. The instance plugin diffs it against the stored list and sends back the names to add and remove instead of the full list. Lua rebuilds the full list from the snapshot kept in the download record and runs the existing deserializer, so the end state is unchanged.

Also stop the serializer from appending to the stored failed recipe list in place, which grew the list on every upload without a download in between.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* Test the inventory sync download handler

Covers the empty response, the full list without a snapshot, the delta with a snapshot, and the IPC registration.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* Apply recipe notification deltas directly in Lua

Review feedback: the download field now always carries the delta and the deserializer applies it against the current state instead of rebuilding the full list first. The previous failed recipe names are passed in so they survive the round trip. The version check moved to the call site.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* Fix failed not being cleared

* Send only the local state on download init

The current state sent with the download request no longer includes the failed names from the previous deserialization. Stored names unknown on this instance always end up in the add list, so the deserializer recomputes the failed list from that on every download and needs no previous state.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* Convert inventory sync test to ES module

Master migrated to ESM after this branch was created, so the CommonJS test failed to load on the merge commit.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com>

plugins/inventory_sync/README.md

@@ -32,8 +32,8 @@ The greatly simplified data flow is as follows:
     3. If the player inventory was previously uploaded but is not the same then the player inventory is deleted and the player is turned into a spectator.
     4. Otherwise the player inventory is kept.
 
-5. Scenario script asks for the player inventory from the controller.
-6. Controller sends it to the scenario in a stream of chunks.
+5. Scenario script asks for the player inventory from the controller, including its current recipe notification state.
+6. Controller sends it to the scenario in a stream of chunks, with only the recipe notifications that differ from the current state.
 7. Scenario displays and updates a progress bar as each chunk is received.
 8. Once all chunks have loaded the player's synced inventory is recreated from the data and the player can start playing.
 

plugins/inventory_sync/instance.ts

@@ -1,3 +1,5 @@
+import util from "util";
+import zlib from "zlib";
 import * as lib from "@clusterio/lib";
 import { BaseInstancePlugin } from "@clusterio/host";
 import {
@@ -8,6 +10,11 @@ type IpcPlayerName = {
 	player_name: string
 }
 
+type IpcDownloadRequest = {
+	player_name: string,
+	recipe_notifications?: string,
+}
+
 type IpcAcquireResponse = {
 	player_name: string,
 	status: string,
@@ -26,6 +33,49 @@ function chunkify(chunkSize: number, string: string): string[] {
 	return string.match(new RegExp(`.{1,${chunkSize}}`, "g")) || [];
 }
 
+const inflate = util.promisify(zlib.inflate);
+const deflate = util.promisify(zlib.deflate);
+
+/**
+ * Decode a string produced by helpers.encode_string in Factorio
+ */
+async function decodeLuaString(encoded: string): Promise<string> {
+	return (await inflate(Buffer.from(encoded, "base64"))).toString("utf8");
+}
+
+/**
+ * Encode a string so that helpers.decode_string in Factorio can read it
+ */
+async function encodeLuaString(text: string): Promise<string> {
+	return (await deflate(Buffer.from(text, "utf8"))).toString("base64");
+}
+
+export type RecipeNotificationDelta = {
+	add?: string[],
+	remove?: string[],
+}
+
+/**
+ * Compute the changes needed to turn the current cleared recipe list into the stored one
+ * @param stored - Cleared recipe names stored on the controller
+ * @param current - Cleared recipe names the instance currently has
+ * @returns names to add to and remove from the current list
+ */
+export function recipeNotificationDelta(stored: string[], current: string[]): RecipeNotificationDelta {
+	const storedSet = new Set(stored);
+	const currentSet = new Set(current);
+	const delta: RecipeNotificationDelta = {};
+	const add = stored.filter(name => !currentSet.has(name));
+	const remove = current.filter(name => !storedSet.has(name));
+	if (add.length) {
+		delta.add = add;
+	}
+	if (remove.length) {
+		delta.remove = remove;
+	}
+	return delta;
+}
+
 export class InstancePlugin extends BaseInstancePlugin {
 	playersToRelease!: Set<string>;
 	disconnecting!: boolean;
@@ -55,7 +105,7 @@ export class InstancePlugin extends BaseInstancePlugin {
 		);
 		this.instance.server.on(
 			"ipc-inventory_sync_download",
-			(request: IpcPlayerName) => this.handleDownload(request).catch(
+			(request: IpcDownloadRequest) => this.handleDownload(request).catch(
 				err => this.logger.error(`Error handling ipc-inventory_sync_download:\n${err.stack}`)
 			),
 		);
@@ -163,7 +213,27 @@ export class InstancePlugin extends BaseInstancePlugin {
 		);
 	}
 
-	async handleDownload(request: IpcPlayerName) {
+	/**
+	 * Replace the stored recipe notifications with the difference from the
+	 * current state on the instance to reduce the amount of data sent to Lua.
+	 */
+	async applyRecipeNotificationDelta(playerData: IpcPlayerData, current?: string) {
+		if (!playerData.recipe_notifications) {
+			return;
+		}
+		try {
+			const delta = recipeNotificationDelta(
+				JSON.parse(await decodeLuaString(playerData.recipe_notifications)),
+				current ? JSON.parse(await decodeLuaString(current)) : [],
+			);
+			playerData.recipe_notifications = await encodeLuaString(JSON.stringify(delta));
+		} catch (err: any) {
+			this.logger.warn(`Dropping invalid recipe notifications for ${playerData.name}:\n${err.stack}`);
+			delete playerData.recipe_notifications;
+		}
+	}
+
+	async handleDownload(request: IpcDownloadRequest) {
 		const playerName = request.player_name;
 		this.logger.verbose(`Downloading ${playerName}`);
 
@@ -177,6 +247,8 @@ export class InstancePlugin extends BaseInstancePlugin {
 			return;
 		}
 
+		await this.applyRecipeNotificationDelta(response.playerData, request.recipe_notifications);
+
 		const chunkSize = this.instance.config.get("inventory_sync.rcon_chunk_size");
 		const chunks = chunkify(chunkSize, JSON.stringify(response.playerData));
 		this.logger.verbose(`Sending inventory for ${playerName} in ${chunks.length} chunks`);

plugins/inventory_sync/module/inventory_sync.lua

@@ -17,6 +17,7 @@ local progress_dialog = require("modules/inventory_sync/gui/progress_dialog")
 local dialog_failed_download = require("modules/inventory_sync/gui/dialog_failed_download")
 
 local v2_remote_controller = compat.version_ge("2.0.0")
+local recipe_notifications_api = compat.version_ge("2.0.67")
 
 -- Returns true if the player is currently in a cutscene
 local function is_in_cutscene(player)
@@ -441,8 +442,15 @@ function inventory_sync.initiate_inventory_download(player, player_record, gener
 	local script_data = get_script_data()
 	script_data.active_downloads[player.name] = record
 
+	-- The plugin only sends back what differs from the current notification state
+	local recipe_notifications
+	if recipe_notifications_api then
+		recipe_notifications = serialize.serialize_crafting_notifications(player)
+	end
+
 	clusterio_api.send_json("inventory_sync_download", {
-		player_name = player.name
+		player_name = player.name,
+		recipe_notifications = recipe_notifications,
 	})
 
 	-- If this is a synced player turn them into a spectator while the

plugins/inventory_sync/module/serialize.lua

@@ -533,62 +533,55 @@ function serialize.serialize_crafting_notifications(player, failed)
 		recipes[recipe.name] = nil
 	end
 
-	-- Convert to a list of names
-	local recipe_names = failed or {} -- Includes previous recipes if any
-	local index = #recipe_names + 1
+	-- Convert to a list of names, including previously failed recipes if any
+	local recipe_names = {}
+	for _, name in ipairs(failed or {}) do
+		recipe_names[#recipe_names + 1] = name
+	end
 	for name in pairs(recipes) do
-		recipe_names[index] = name
-		index = index + 1
+		recipe_names[#recipe_names + 1] = name
 	end
 
 	return helpers.encode_string(helpers.table_to_json(recipe_names))
 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
+
 --- @param player LuaPlayer
---- @param serialized string
+--- @param serialized string Encoded delta against the state sent with the download request
 --- @return string[]? failed Recipe names that do not exist
 function serialize.deserialize_crafting_notifications(player, serialized)
-	-- Decode the recipe name strings
-	local recipe_names = helpers.json_to_table(assert(helpers.decode_string(serialized)))
-	assert(type(recipe_names) == "table", "wrong type decoded from json_to_table")
-	local clear_notification = player.clear_recipe_notification
-	local add_notification = player.add_recipe_notification
-	local notifications = {}
-	local failed = {}
+	--- @type CraftingNotificationDelta
+	local delta = helpers.json_to_table(assert(helpers.decode_string(serialized)))
+	assert(type(delta) == "table", "wrong type decoded from json_to_table")
 
-	-- Assume all recipes need a notification
+	-- Recipes which can have a notification on this server
+	local enabled = {}
 	for _, recipe in pairs(player.force.recipes) do
 		if recipe.enabled and not recipe.hidden then
-			notifications[recipe.name] = true
+			enabled[recipe.name] = true
 		end
 	end
 
-	-- Remove recipes the player has cleared
-	for _, recipe_name in pairs(recipe_names) do
-		if notifications[recipe_name] then
-			notifications[recipe_name] = false
+	-- Clear notifications the player has cleared elsewhere, this means "add seen notification"
+	local failed = {}
+	for _, recipe_name in pairs(delta.add or {}) do
+		if enabled[recipe_name] then
+			player.clear_recipe_notification(recipe_name)
 		else
-			-- Recipe does not exist on this server
-			failed[#failed+1] = recipe_name
+			failed[#failed + 1] = recipe_name -- Recipe does not exist on this server
 		end
 	end
 
-	-- Clear ones which are ones which are not required
-	for _, recipe in pairs(player.get_recipe_notifications()) do
-		if notifications[recipe.name] == false then
-			clear_notification(recipe.name)
-		end
-		notifications[recipe.name] = nil -- Prevents double add below
-	end
-
-	-- Add any which the player did not have
-	for recipe_name, should_add in pairs(notifications) do
-		if should_add then
-			add_notification(recipe_name)
+	-- Add notifications the player has elsewhere, this means "remove seen notification"
+	for _, recipe_name in pairs(delta.remove or {}) do
+		if enabled[recipe_name] then
+			player.add_recipe_notification(recipe_name)
 		end
 	end
 
-	-- Return failed recipes if any
 	return next(failed) and failed or nil
 end
 

plugins/inventory_sync/test/instance.js

@@ -0,0 +1,130 @@
+import assert from "node:assert/strict";
+import util from "node:util";
+import zlib from "node:zlib";
+import * as lib from "@clusterio/lib";
+
+import * as mock from "../../../test/mock.js";
+import { recipeNotificationDelta, InstancePlugin } from "../dist/node/instance.js";
+import { DownloadResponse } from "../dist/node/messages.js";
+import { plugin as info } from "../dist/node/index.js";
+
+const inflate = util.promisify(zlib.inflate);
+const deflate = util.promisify(zlib.deflate);
+
+// Matches helpers.encode_string in Factorio
+async function encode(value) {
+	return (await deflate(Buffer.from(JSON.stringify(value)))).toString("base64");
+}
+async function decode(encoded) {
+	return JSON.parse((await inflate(Buffer.from(encoded, "base64"))).toString());
+}
+
+describe("inventory_sync", function() {
+	describe("recipeNotificationDelta()", function() {
+		it("should be empty when the lists match", function() {
+			assert.deepEqual(recipeNotificationDelta(["a", "b"], ["b", "a"]), {});
+		});
+		it("should add stored names missing from current", function() {
+			assert.deepEqual(recipeNotificationDelta(["a", "b", "c"], ["a"]), { add: ["b", "c"] });
+		});
+		it("should remove current names missing from stored", function() {
+			assert.deepEqual(recipeNotificationDelta(["a"], ["a", "b"]), { remove: ["b"] });
+		});
+		it("should handle both directions at once", function() {
+			assert.deepEqual(recipeNotificationDelta(["a", "b"], ["b", "c"]), { add: ["a"], remove: ["c"] });
+		});
+		it("should keep duplicates in stored out of the current list", function() {
+			assert.deepEqual(recipeNotificationDelta(["a", "a"], ["a"]), {});
+		});
+	});
+
+	describe("InstancePlugin.applyRecipeNotificationDelta()", function() {
+		const warnings = [];
+		const plugin = { logger: { warn: msg => warnings.push(msg) } };
+		const apply = InstancePlugin.prototype.applyRecipeNotificationDelta.bind(plugin);
+
+		it("should replace the stored list with an encoded delta", async function() {
+			const playerData = { name: "test", recipe_notifications: await encode(["a", "b"]) };
+			await apply(playerData, await encode(["b", "c"]));
+			assert.deepEqual(await decode(playerData.recipe_notifications), { add: ["a"], remove: ["c"] });
+		});
+		it("should send an empty delta when nothing changed", async function() {
+			const playerData = { name: "test", recipe_notifications: await encode(["a"]) };
+			await apply(playerData, await encode(["a"]));
+			assert.deepEqual(await decode(playerData.recipe_notifications), {});
+		});
+		it("should add everything 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"] });
+		});
+		it("should do nothing without stored notifications", async function() {
+			const playerData = { name: "test" };
+			await apply(playerData, await encode(["a"]));
+			assert.deepEqual(playerData, { name: "test" });
+		});
+		it("should drop the field on invalid input", async function() {
+			const playerData = { name: "test", recipe_notifications: "not base64 zlib" };
+			await apply(playerData, await encode(["a"]));
+			assert.equal(playerData.recipe_notifications, undefined);
+			assert.equal(warnings.length, 1);
+		});
+	});
+
+	describe("InstancePlugin.handleDownload()", function() {
+		let instancePlugin;
+		before(async function() {
+			try {
+				lib.registerPluginMessages([info]);
+			} catch (err) {
+				// Already registered by the full test suite
+			}
+			instancePlugin = await mock.createInstancePlugin(InstancePlugin, info);
+			instancePlugin.instance.mockConfigEntries.set("inventory_sync.rcon_chunk_size", 100000);
+		});
+		beforeEach(function() {
+			instancePlugin.instance.server.reset();
+		});
+
+		function respondWith(playerData) {
+			instancePlugin.instance.connector.once("send", message => {
+				instancePlugin.instance.connector.emit("message", new lib.MessageResponse(
+					1, message.dst, message.src, new DownloadResponse(playerData)
+				));
+			});
+		}
+		function downloadCommand(playerData) {
+			const json = lib.escapeString(JSON.stringify(playerData));
+			return `/sc inventory_sync.download_inventory('test','${json}',1,1)`;
+		}
+
+		it("should send an empty download when the controller has no data", async function() {
+			respondWith(undefined);
+			await instancePlugin.handleDownload({ player_name: "test" });
+			assert.deepEqual(
+				instancePlugin.instance.server.rconCommands,
+				["/sc inventory_sync.download_inventory('test',nil,0,0)"]
+			);
+		});
+		it("should send everything as a delta when no snapshot was given", async function() {
+			respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a"]) });
+			await instancePlugin.handleDownload({ player_name: "test" });
+			const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) };
+			assert.deepEqual(instancePlugin.instance.server.rconCommands, [downloadCommand(expected)]);
+		});
+		it("should send a delta when a snapshot was given", async function() {
+			respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a", "b"]) });
+			await instancePlugin.handleDownload({ player_name: "test", recipe_notifications: await encode(["b"]) });
+			const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) };
+			assert.deepEqual(instancePlugin.instance.server.rconCommands, [downloadCommand(expected)]);
+		});
+		it("should be invoked by the download ipc", async function() {
+			respondWith(undefined);
+			instancePlugin.instance.server.emit("ipc-inventory_sync_download", { player_name: "test" });
+			for (let i = 0; i < 100 && !instancePlugin.instance.server.rconCommands.length; i++) {
+				await new Promise(resolve => setImmediate(resolve));
+			}
+			assert.equal(instancePlugin.instance.server.rconCommands.length, 1);
+		});
+	});
+});