From 5589609b1b4ebdb082323772811c6fa222220e4e Mon Sep 17 00:00:00 2001 From: bbassie <17990055+bbassie@users.noreply.github.com> Date: Fri, 30 Oct 2026 00:04:15 +0000 Subject: [PATCH] Resync research after a new controller session While the host has no controller session, events between the instance and the controller are dropped in both directions, and research_sync only synced technologies on instance start. A technology finished during the outage stayed unresearched on the controller and on every other instance until the instance restarted. On a new session the instance now runs the same technology sync as on start, before taking the cluster queue. Contributions made while the controller is unreachable are summed per technology and sent on connect or resume, since progress from several instances can't be rebuilt from a state sync that keeps the highest progress. A finish made while the connection was dropped triggers the sync on resume too. sync_technologies now runs while the game is running, so it skips writing progress to a technology that finished after the dump. Fixes #1123 Co-Authored-By: Claude Opus 5.5 --- plugins/research_sync/instance.ts | 124 +++++++++++++++-------- plugins/research_sync/module/sync.lua | 3 +- plugins/research_sync/test/controller.js | 49 +++++++++ plugins/research_sync/test/instance.js | 114 ++++++++++++++++++++- 4 files changed, 246 insertions(+), 44 deletions(-) diff --git a/plugins/research_sync/instance.ts b/plugins/research_sync/instance.ts index 02771abf..464e91ea 100644 --- a/plugins/research_sync/instance.ts +++ b/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(); + let finishedWhileDisconnected = false; // Commands are sent one at a time so they execute in order let rconQueue: Promise = 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); } diff --git a/plugins/research_sync/module/sync.lua b/plugins/research_sync/module/sync.lua index 65809005..3b675c43 100644 --- a/plugins/research_sync/module/sync.lua +++ b/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) diff --git a/plugins/research_sync/test/controller.js b/plugins/research_sync/test/controller.js index faef14c5..69584a15 100644 --- a/plugins/research_sync/test/controller.js +++ b/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(); diff --git a/plugins/research_sync/test/instance.js b/plugins/research_sync/test/instance.js index 844d16f2..64138a78 100644 --- a/plugins/research_sync/test/instance.js +++ b/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);