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