Merge pull request #1155 from bbassie/fix/research-sync-session-loss

Resync research_sync after a new controller session

plugins/research_sync/instance.ts

@@ -24,9 +24,13 @@ type IpcFinished = {
 type IpcQueue = { queue: unknown, player?: string };
 
 export default async function(context: InstancePluginContext) {
-	const { instance, logger, plugin } = context;
+	const { instance, host, logger, plugin } = context;
 	let syncStarted = false;
 	let queueEventsApplied = 0;
+	// Contributions made while the controller is unreachable, summed per technology.
+	// Unlike finished research they can't be recovered from the game state.
+	const pendingContributions = new Map<string, IpcContribution>();
+	let finishedWhileDisconnected = false;
 
 	// Commands are sent one at a time so they execute in order
 	let rconQueue: Promise<unknown> = Promise.resolve();
@@ -69,11 +73,68 @@ export default async function(context: InstancePluginContext) {
 	}
 
 	async function researchContribution(tech: IpcContribution) {
-		instance.sendTo("controller", new ContributionEvent(tech.name, tech.level, tech.contribution));
+		if (host.connector.connected) {
+			instance.sendTo("controller", new ContributionEvent(tech.name, tech.level, tech.contribution));
+			return;
+		}
+		let pending = pendingContributions.get(tech.name);
+		if (pending && pending.level === tech.level) {
+			pending.contribution += tech.contribution;
+		} else if (!pending || pending.level < tech.level) {
+			// A lower level finished here, the technology sync reports it
+			pendingContributions.set(tech.name, { ...tech });
+		}
 	}
 
 	async function researchFinished(tech: IpcFinished) {
-		instance.sendTo("controller", new FinishedEvent(tech.name, tech.level));
+		if (host.connector.connected) {
+			instance.sendTo("controller", new FinishedEvent(tech.name, tech.level));
+			return;
+		}
+		finishedWhileDisconnected = true;
+	}
+
+	function sendPendingContributions() {
+		for (let tech of pendingContributions.values()) {
+			instance.sendTo("controller", new ContributionEvent(tech.name, tech.level, tech.contribution));
+		}
+		pendingContributions.clear();
+	}
+
+	async function syncTechnologies() {
+		let dumpJson = await sendOrderedRcon("/sc research_sync.dump_technologies()");
+		let techsToSend = [];
+		let instanceTechs = new Map();
+		for (let tech of JSON.parse(dumpJson)) {
+			techsToSend.push(new TechnologySync(
+				tech.name,
+				tech.level,
+				tech.progress || null,
+				tech.researched,
+			));
+			instanceTechs.set(tech.name, tech);
+		}
+
+		let controllerTechs = await instance.sendTo("controller", new SyncTechnologiesRequest(techsToSend));
+		syncStarted = true;
+		let techsToSync = [];
+		for (let controllerTech of controllerTechs) {
+			let { name, level, progress, researched } = controllerTech;
+			let instanceTech = instanceTechs.get(name);
+			if (
+				!instanceTech
+				|| instanceTech.level !== level
+				|| (instanceTech.progress || null) !== progress
+				|| instanceTech.researched !== researched
+			) {
+				techsToSync.push(controllerTech);
+			}
+		}
+
+		if (techsToSync.length) {
+			let syncJson = lib.escapeString(JSON.stringify(techsToSync));
+			await sendOrderedRcon(`/sc research_sync.sync_technologies("${syncJson}")`, true);
+		}
 	}
 
 	instance.server.on("ipc-research_sync:contribution", (tech: IpcContribution) => {
@@ -128,52 +189,35 @@ export default async function(context: InstancePluginContext) {
 	});
 
 	instance.hooks.start.attach(plugin.name, async () => {
-		let dumpJson = await sendOrderedRcon("/sc research_sync.dump_technologies()");
-		let techsToSend = [];
-		let instanceTechs = new Map();
-		for (let tech of JSON.parse(dumpJson)) {
-			techsToSend.push(new TechnologySync(
-				tech.name,
-				tech.level,
-				tech.progress || null,
-				tech.researched,
-			));
-			instanceTechs.set(tech.name, tech);
-		}
-
-		let controllerTechs = await instance.sendTo("controller", new SyncTechnologiesRequest(techsToSend));
-		syncStarted = true;
-		let techsToSync = [];
-		for (let controllerTech of controllerTechs) {
-			let { name, level, progress, researched } = controllerTech;
-			let instanceTech = instanceTechs.get(name);
-			if (
-				!instanceTech
-				|| instanceTech.level !== level
-				|| (instanceTech.progress || null) !== progress
-				|| instanceTech.researched !== researched
-			) {
-				techsToSync.push(controllerTech);
-			}
-		}
-
-		if (techsToSync.length) {
-			let syncJson = lib.escapeString(JSON.stringify(techsToSync));
-			await sendOrderedRcon(`/sc research_sync.sync_technologies("${syncJson}")`, true);
-		}
+		await syncTechnologies();
 
 		// A first start counts as turning following on, which seeds an empty cluster queue
 		await syncQueue(true);
 	});
 
-	// QueueEvents sent while the controller session was down are lost, so take the cluster queue again.
-	// An edit made on this instance during the outage was never sent and is replaced: the cluster queue wins.
+	// Events are lost both ways while the controller session is down. Contributions are kept above, finished
+	// research comes from the technology sync, and the cluster queue replaces queue edits made meanwhile.
 	instance.hooks.controllerConnectionEvent.attach(plugin.name, async (event) => {
-		if (event !== "connect" || !syncStarted || !["starting", "running"].includes(instance.status)) {
+		if (event !== "connect" && event !== "resume") {
 			return;
 		}
 		try {
-			await syncQueue(false);
+			sendPendingContributions();
+			if (!syncStarted || !["starting", "running"].includes(instance.status)) {
+				return;
+			}
+			if (event === "connect" || finishedWhileDisconnected) {
+				finishedWhileDisconnected = false;
+				try {
+					await syncTechnologies();
+				} catch (err) {
+					finishedWhileDisconnected = true;
+					throw err;
+				}
+			}
+			if (event === "connect") {
+				await syncQueue(false);
+			}
 		} catch (err: any) {
 			unexpectedError(err);
 		}

plugins/research_sync/module/sync.lua

@@ -332,7 +332,8 @@ function research_sync.sync_technologies(data)
 				end
 				tech.researched = true
 				progress = nil
-			elseif tech_data[progressIndex] then
+			-- may have finished here since the dump this sync is answering
+			elseif tech_data[progressIndex] and not tech.researched then
 				send_contribution(tech)
 				progress = tech_data[progressIndex]
 				set_technology_progress(tech, progress)

plugins/research_sync/test/controller.js

@@ -164,6 +164,55 @@ describe("research_sync controller", function() {
 		});
 	});
 
+	// What instances send on reconnecting: the contributions summed during the outage, then a technology sync
+	describe("reconnect", function() {
+		const contribute = (name, level, contribution) => mock.getHandler(controller, ContributionEvent)(
+			new ContributionEvent(name, level, contribution), src
+		);
+		const syncTechs = async (...techs) => new Map(
+			(await mock.getHandler(controller, SyncTechnologiesRequest)(new SyncTechnologiesRequest(techs), src))
+				.map(tech => [tech.name, tech])
+		);
+		const finishedEvents = () => sent.filter(([, m]) => m instanceof FinishedEvent).map(([, m]) => m);
+
+		it("should count progress made on every instance during the outage", async function() {
+			await contribute("automation", 1, 0.25);
+			// A and B each contributed 0.25 from the controller's 0.25
+			await contribute("automation", 1, 0.25);
+			await syncTechs(new TechnologySync("automation", 1, 0.5, false));
+			await contribute("automation", 1, 0.25);
+			let techs = await syncTechs(new TechnologySync("automation", 1, 0.5, false));
+			assert.equal(techs.get("automation").progress, 0.75);
+		});
+		it("should broadcast research finished during the outage once", async function() {
+			await contribute("automation", 1, 0.5);
+			await syncTechs(new TechnologySync("automation", 1, null, true));
+			await syncTechs(new TechnologySync("automation", 1, null, true));
+			assert.deepEqual(finishedEvents(), [new FinishedEvent("automation", 1)]);
+		});
+		const tech = (level, progress) => new TechnologySync("mining-productivity-3", level, progress, false);
+
+		it("should take a higher infinite research level from an instance", async function() {
+			await contribute("mining-productivity-3", 4, 0.9);
+			let techs = await syncTechs(tech(5, 0.2));
+			assert.deepEqual(finishedEvents(), [new FinishedEvent("mining-productivity-3", 4)]);
+			assert.deepEqual(techs.get("mining-productivity-3"), tech(5, 0.2));
+		});
+		it("should return a higher infinite research level to an instance", async function() {
+			await syncTechs(tech(6, 0.1));
+			sent.length = 0;
+			let techs = await syncTechs(tech(4, 0.5));
+			assert.equal(finishedEvents().length, 0);
+			assert.deepEqual(techs.get("mining-productivity-3"), tech(6, 0.1));
+		});
+		it("should ignore contributions to a level finished elsewhere", async function() {
+			await syncTechs(tech(5, null));
+			await contribute("mining-productivity-3", 4, 0.5);
+			let techs = await syncTechs();
+			assert.deepEqual(techs.get("mining-productivity-3"), tech(5, null));
+		});
+	});
+
 	it("should persist the queue across loads", async function() {
 		await update(["a", "b"]);
 		await controller.hooks.save.invoke();

plugins/research_sync/test/instance.js

@@ -5,11 +5,12 @@ import * as mock from "../../../test/mock.js";
 import loadInstancePlugin from "../dist/node/instance.js";
 import { plugin as info } from "../dist/node/index.js";
 import {
-	QueueEvent, QueueSyncRequest, QueueUpdateEvent, SyncTechnologiesRequest,
+	ContributionEvent, FinishedEvent, QueueEvent, QueueSyncRequest, QueueUpdateEvent, SyncTechnologiesRequest,
+	TechnologySync,
 } from "../dist/node/messages.js";
 
 describe("research_sync instance", function() {
-	let instance, sent;
+	let instance, host, sent;
 	const clusterQueue = ["a", "b"];
 
 	before(function() {
@@ -20,7 +21,7 @@ describe("research_sync instance", function() {
 		}
 	});
 	beforeEach(async function() {
-		({ instance } = await mock.loadInstancePlugin(loadInstancePlugin, info));
+		({ instance, host } = await mock.loadInstancePlugin(loadInstancePlugin, info));
 		sent = [];
 		instance.sendTo = async (dst, message) => {
 			sent.push(message);
@@ -95,6 +96,7 @@ describe("research_sync instance", function() {
 			await instance.hooks.start.invoke();
 			sent.length = 0;
 			instance.server.reset();
+			instance.server.rconCommandResults.set("/sc research_sync.dump_technologies()", "[]");
 			queueState({ follow: true, first: false });
 		});
 
@@ -108,6 +110,31 @@ describe("research_sync instance", function() {
 			await instance.hooks.controllerConnectionEvent.invoke("resume");
 			assert.equal(sent.length, 0);
 		});
+		it("should sync technologies before the queue on a new session", async function() {
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+			assert.ok(sent[0] instanceof SyncTechnologiesRequest);
+			assert.ok(sent[1] instanceof QueueSyncRequest);
+		});
+		it("should apply research finished elsewhere during the outage", async function() {
+			instance.server.rconCommandResults.set(
+				"/sc research_sync.dump_technologies()",
+				JSON.stringify([{ name: "automation", level: 1, progress: 0.5, researched: false }]),
+			);
+			let controllerTechs = [new TechnologySync("automation", 1, null, true)];
+			let sendTo = instance.sendTo;
+			instance.sendTo = async (dst, message) => {
+				if (message instanceof SyncTechnologiesRequest) {
+					sent.push(message);
+					return controllerTechs;
+				}
+				return sendTo(dst, message);
+			};
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+			let syncRequest = sent.find(m => m instanceof SyncTechnologiesRequest);
+			assert.deepEqual(syncRequest.technologies, [new TechnologySync("automation", 1, 0.5, false)]);
+			let syncJson = lib.escapeString(JSON.stringify(controllerTechs));
+			assert.ok(instance.server.rconCommands.includes(`/sc research_sync.sync_technologies("${syncJson}")`));
+		});
 		it("should do nothing while stopped", async function() {
 			instance.status = "stopped";
 			await instance.hooks.controllerConnectionEvent.invoke("connect");
@@ -122,6 +149,9 @@ describe("research_sync instance", function() {
 			// reply resolves the request's await
 			instance.sendTo = async (dst, message) => {
 				sent.push(message);
+				if (message instanceof SyncTechnologiesRequest) {
+					return [];
+				}
 				mock.getHandler(instance, QueueEvent)(new QueueEvent(["new"], null, "B"));
 				return ["old"];
 			};
@@ -130,6 +160,84 @@ describe("research_sync instance", function() {
 		});
 	});
 
+	describe("controller outage", function() {
+		beforeEach(async function() {
+			queueState({ follow: false });
+			await instance.hooks.start.invoke();
+			sent.length = 0;
+			host.connector.connected = false;
+		});
+		const contributions = () => sent.filter(m => m instanceof ContributionEvent);
+		const techSyncs = () => sent.filter(m => m instanceof SyncTechnologiesRequest);
+		function contribute(name, level, contribution) {
+			instance.server.emit("ipc-research_sync:contribution", { name, level, contribution });
+		}
+
+		it("should send contributions while connected", async function() {
+			host.connector.connected = true;
+			contribute("automation", 1, 0.25);
+			instance.server.emit("ipc-research_sync:finished", { name: "automation", level: 1 });
+			assert.ok(sent[0] instanceof ContributionEvent);
+			assert.ok(sent[1] instanceof FinishedEvent);
+		});
+		it("should sum contributions and send them on a new session", async function() {
+			contribute("automation", 1, 0.25);
+			contribute("automation", 1, 0.25);
+			contribute("logistics", 1, 0.125);
+			assert.equal(sent.length, 0);
+			host.connector.connected = true;
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+			assert.deepEqual(
+				contributions().map(m => [m.name, m.level, m.contribution]),
+				[["automation", 1, 0.5], ["logistics", 1, 0.125]],
+			);
+			assert.ok(sent.indexOf(contributions()[1]) < sent.indexOf(techSyncs()[0]));
+		});
+		it("should keep only contributions to the highest level", async function() {
+			contribute("mining-productivity-3", 3, 0.5);
+			contribute("mining-productivity-3", 4, 0.25);
+			contribute("mining-productivity-3", 3, 0.5);
+			host.connector.connected = true;
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.deepEqual(
+				contributions().map(m => [m.name, m.level, m.contribution]),
+				[["mining-productivity-3", 4, 0.25]],
+			);
+		});
+		it("should send contributions once", async function() {
+			contribute("automation", 1, 0.25);
+			host.connector.connected = true;
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.equal(contributions().length, 1);
+		});
+		it("should not sync technologies on resume when nothing finished", async function() {
+			contribute("automation", 1, 0.25);
+			host.connector.connected = true;
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.equal(techSyncs().length, 0);
+		});
+		it("should sync technologies on resume when research finished", async function() {
+			instance.server.emit("ipc-research_sync:finished", { name: "automation", level: 1 });
+			assert.equal(sent.length, 0);
+			host.connector.connected = true;
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.equal(techSyncs().length, 1);
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.equal(techSyncs().length, 1);
+		});
+		it("should retry the technology sync after it fails on resume", async function() {
+			instance.server.emit("ipc-research_sync:finished", { name: "automation", level: 1 });
+			host.connector.connected = true;
+			let sendTo = instance.sendTo;
+			instance.sendTo = async () => { throw new Error("gone"); };
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			instance.sendTo = sendTo;
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.equal(techSyncs().length, 1);
+		});
+	});
+
 	it("should forward queue changes from the game", async function() {
 		instance.server.emit("ipc-research_sync:queue", { queue: {}, player: "Bob" });
 		assert.equal(sent.length, 1);