diff --git a/src/m2d/event-dispatcher.js b/src/m2d/event-dispatcher.js index c71f574..73faba0 100644 --- a/src/m2d/event-dispatcher.js +++ b/src/m2d/event-dispatcher.js @@ -130,7 +130,7 @@ async function sendError(roomID, source, type, e, payload) { || (e instanceof mreq.MatrixServerError && [502, 503].includes(e.httpStatus) && e.errcode === "CX_SERVER_ERROR") ) { // Matrix homeserver is down (reverse proxy indicated failure) - homeserverStatus.homeserverStatus.setErrorWithPacket([source, payload]) + homeserverStatus.homeserverStatus.queuePacket([source, payload]) } } } diff --git a/src/matrix/homeserver-status.js b/src/matrix/homeserver-status.js index c50d5ef..1f6db93 100644 --- a/src/matrix/homeserver-status.js +++ b/src/matrix/homeserver-status.js @@ -1,8 +1,8 @@ // @ts-check -const assert = require("assert").strict const Denque = require("denque") const StateMachine = require("snowtransfer").StateMachine +const Ty = require("../types") const passthrough = require("../passthrough") const {sync} = passthrough @@ -19,7 +19,7 @@ function debugHomeserverStatus(message) { } } -const homeserverStatus = sync.remember(() => new class HomeserverStatus { +class HomeserverStatus { constructor() { /** @private */ this.queue = new Denque() @@ -39,10 +39,14 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { .defineState("checking", { onEnter: [async (event) => { - this.checkingWhileOnline = event === "check while online" + this.checkingWhileOnline = event === "check assume online" const pingResult = await api.ping().catch(e => ({ok: false})) this.lastCheck = Date.now() - if (pingResult.ok) { + // ping result is stale if something else changed the state in the async gap + if (this.sm.currentStateName !== "checking") return + if (pingResult.ok && !this.queue.isEmpty()) { + this.sm.doTransition("check ok with queue") + } else if (pingResult.ok) { this.sm.doTransition("check ok") } else { this.sm.doTransition("check fail") @@ -58,6 +62,8 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { this.pingInterval = setInterval(async () => { const pingResult = await api.ping().catch(e => ({ok: false, status: "net", root: e.message})) this.lastCheck = Date.now() + // ping result is stale if something else changed the state in the async gap + if (this.sm.currentStateName !== "offline") return if (pingResult.ok) { this.sm.doTransition("ping ok") } @@ -76,11 +82,16 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { while (!this.queue.isEmpty()) { const [source, packet] = this.queue.peekFront() // same position as .shift() debugHomeserverStatus(`homeserver status: ${new Date().toISOString()} dq sourceAndPacket ${packet.t} ${packet.d?.content}`) - if (source === "Discord") { - await discordPackets.dispatchPacketToBridge(passthrough.discord, packet) - } else { - // Unfortunately cannot await this, just have to trust it - passthrough.as.emit(`type:${packet.type}`, packet) + try { + if (source === "Discord") { + await discordPackets.dispatchPacketToBridge(passthrough.discord, packet) + } else { + await Promise.all(passthrough.as.listeners(`type:${packet.type}`).map(listener => listener(packet))) + } + } catch (e) { + // `dispatchPacketToBridge` should not throw, this is here in case of dev error. + // Skip the packet so that it can't block the queue forever. + console.error("[homeserver status] Skipping a queued event that couldn't be replayed:", e) } if (this.sm.currentStateName !== "recovering") return // got kicked out due to another error this.queue.shift() @@ -99,6 +110,7 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { .defineTransition("online", "check assume online", "checking") .defineTransition("online", "check assume offline", "checking") .defineTransition("checking", "check ok", "online") + .defineTransition("checking", "check ok with queue", "recovering") .defineTransition("checking", "check fail", "offline") this.sm.on("enter", st => debugHomeserverStatus(`homeserver status: ${st}`)) @@ -137,20 +149,28 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { } /** - * When offline or recovering, call this for incoming packets to queue them to be sent in order later. + * Queue a packet to be sent in order later, particularly when the homeserver is not realtime. + * + * Queueing while realtime implies the homeserver went down, so this also transitions the state to offline. + * + * @param {["Discord", import("cloudstorm").IGatewayMessage] + * | ["Matrix", Ty.Event.Outer]} sourceAndPacket */ queuePacket(sourceAndPacket) { - assert(["offline", "recovering"].includes(this.sm.currentStateName)) + if (this.sm.currentStateName === "recovering" && this.queue.peekFront()?.[1] === sourceAndPacket[1]) { + this.sm.doTransition("error") + return + } + + if (this.isRealTime()) { + this.sm.doTransition("error") + } + this.queue.push(sourceAndPacket) } +} - setErrorWithPacket(sourceAndPacket) { - const wasRecovering = this.sm.currentStateName === "recovering" - this.sm.doTransition("error") - if (!wasRecovering) { // if was recovering then sourceAndPacket is already in the right place in queue - this.queuePacket(sourceAndPacket) - } - } -}) +const homeserverStatus = sync.remember(() => new HomeserverStatus()) module.exports.homeserverStatus = homeserverStatus +module.exports._HomeserverStatus = HomeserverStatus diff --git a/src/matrix/homeserver-status.test.js b/src/matrix/homeserver-status.test.js new file mode 100644 index 0000000..85a9b58 --- /dev/null +++ b/src/matrix/homeserver-status.test.js @@ -0,0 +1,247 @@ +// @ts-check + +const {test} = require("supertape") +const {EventEmitter, on} = require("events") +const timers = require("timers/promises") +const DiscordTypes = require("discord-api-types/v10") +const Ty = require("../types") +const data = require("../../test/data") +const passthrough = require("../passthrough") +const {sync} = passthrough +/** @type {import("./api")} */ +const api = sync.require("./api") +/** @type {import("../d2m/discord-packets")} */ +const discordPackets = sync.require("../d2m/discord-packets") +const {_HomeserverStatus: HomeserverStatus} = require("./homeserver-status") + +const originalPing = api.ping +const originalDispatch = discordPackets.dispatchPacketToBridge +const originalLog = console.log +const originalError = console.error + +/** + * Create a `MESSAGE_CREATE` discord packet + * @param {string} content + * @returns {import("cloudstorm").IGatewayMessage} + */ +function discordMessage(content) { + return { + op: DiscordTypes.GatewayOpcodes.Dispatch, + t: DiscordTypes.GatewayDispatchEvents.MessageCreate, + s: 1, + d: {...data.message.simple_plaintext, content}, + shard_id: 0 + } +} + +/** + * Create a `m.room.message` matrix event + * @param {string} body + * @returns {Ty.Event.Outer_M_Room_Message} + */ +function matrixMessage(body) { + return { + type: "m.room.message", + room_id: "!kLRqKKUQXcibIMtOpl:cadence.moe", + sender: "@cadence:cadence.moe", + content: {msgtype: "m.text", body}, + origin_server_ts: 0, + event_id: "$event" + } +} + +/** + * Overriding the `HomeserverStatus` type to expose some private items. + * @typedef {Omit & { + * sm: HomeserverStatus["sm"], + * queue: HomeserverStatus["queue"], + * }} ExposedHomeserverStatus + */ + +/** @returns {{hs: ExposedHomeserverStatus, pings: PromiseWithResolvers[]}} */ +function setup() { + console.log = () => {} + console.error = () => {} + const pings = [] + api.ping = () => { + const ping = Promise.withResolvers() + pings.push(ping) + return ping.promise + } + /** @type {ExposedHomeserverStatus} */ // @ts-ignore + const hs = new HomeserverStatus() + return {hs, pings} +} + +/** + * @param {(packet: import("cloudstorm").IGatewayMessage) => any} [onDiscord] called for each replayed Discord packet + * @param {(event: Ty.Event.Outer_M_Room_Message) => any} [onMatrix] called for each replayed m.room.message event + */ +function stubBridge( + onDiscord = () => { }, + onMatrix = () => { }, +) { + discordPackets.dispatchPacketToBridge = async (_client, packet) => onDiscord(packet) + const as = new EventEmitter() + as.on("type:m.room.message", async event => onMatrix(event)) + // @ts-ignore + passthrough.as = as +} + +/** Undo the effects of `setup()` and `stubBridge()` */ +function restore() { + console.log = originalLog + console.error = originalError + api.ping = originalPing + discordPackets.dispatchPacketToBridge = originalDispatch + // @ts-ignore + delete passthrough.as +} + +/** + * Pretend the offline ping succeeded and wait until the end of queue replay. + * @param {import("supertape").Test} t + * @param {ExposedHomeserverStatus} hs + * @param {string} [endState] the state the replay is expected to end in + */ +async function replayQueue(t, hs, endState = "online") { + t.equal(hs.sm.currentStateName, "offline", "replaying the queue from offline") + const enters = on(hs.sm, "enter") // listen before transitioning, since the end state might be reached straight away + hs.sm.doTransition("ping ok") + for await (const [stateName] of enters) { + if (stateName === endState) break + } +} + +test("homeserver status: check assuming online stays real time", async t => { + const {hs, pings} = setup() + const online = hs.waitForOnline(false) + t.equal(hs.sm.currentStateName, "checking") + t.ok(hs.isRealTime()) + + pings[0].resolve({ok: true}) + await online + t.equal(hs.sm.currentStateName, "online") + restore() +}) + +test("homeserver status: check assuming offline replays queue on success", async t => { + const {hs, pings} = setup() + const replayed = [] + stubBridge(packet => replayed.push(packet)) + const online = hs.waitForOnline(true) + t.equal(hs.sm.currentStateName, "checking") + t.notOk(hs.isRealTime()) + + const packet = discordMessage("sent during the check") + hs.queuePacket(["Discord", packet]) + pings[0].resolve({ok: true}) + await online + t.deepEqual(replayed, [packet]) + restore() +}) + +test("homeserver status: successful check ping ignored if packet fails mid-check", async t => { + const {hs, pings} = setup() + stubBridge() + hs.waitForOnline(false) + hs.queuePacket(["Discord", discordMessage("failed during the check")]) + t.equal(hs.sm.currentStateName, "offline") + + // check ping suceeds, but it's too late + pings[0].resolve({ok: true}) + await timers.setImmediate() + t.equal(hs.sm.currentStateName, "offline", "check result was ignored") + await replayQueue(t, hs) + restore() +}) + +test("homeserver status: failed check ping goes offline, replays queue", async t => { + const {hs, pings} = setup() + stubBridge() + const online = hs.waitForOnline(true) + pings[0].resolve({ok: false}) + await timers.setImmediate() + t.equal(hs.sm.currentStateName, "offline") + + await replayQueue(t, hs) + await online + t.equal(hs.sm.currentStateName, "online") + restore() +}) + +test("homeserver status: no new check if the last one was recent", async t => { + const {hs, pings} = setup() + const first = hs.waitForOnline(false) + pings[0].resolve({ok: true}) + await first + + await hs.waitForOnline(false) + t.equal(pings.length, 1, "didn't ping again") + + const forced = hs.waitForOnline(true) + t.equal(pings.length, 2, "forced check pings anyway") + pings[1].resolve({ok: true}) + await forced + restore() +}) + +test("homeserver status: replay waits for matrix events", async t => { + const {hs} = setup() + const slow = matrixMessage("slow") + const next = discordMessage("next") + const replayed = [] + stubBridge( + packet => replayed.push(packet), + async event => { + await timers.setTimeout(10) + replayed.push(event) + } + ) + hs.queuePacket(["Matrix", slow]) + hs.queuePacket(["Discord", next]) + + await replayQueue(t, hs) + t.deepEqual(replayed, [slow, next]) + restore() +}) + +test("homeserver status: replay skips packets that throw", async t => { + const {hs} = setup() + const broken = discordMessage("throws") + const next = discordMessage("next") + const replayed = [] + stubBridge(packet => { + if (packet === broken) throw new Error("this packet is broken") + replayed.push(packet) + }) + hs.queuePacket(["Discord", broken]) + hs.queuePacket(["Discord", next]) + + await replayQueue(t, hs) + t.deepEqual(replayed, [next]) + t.equal(hs.queue.length, 0) + restore() +}) + +test("homeserver status: failing queue packet keeps its queue position", async t => { + const {hs} = setup() + const first = discordMessage("first") + const second = discordMessage("second") + hs.queuePacket(["Discord", first]) + hs.queuePacket(["Discord", second]) + + stubBridge(packet => hs.queuePacket(["Discord", packet])) // homeserver went down again + await replayQueue(t, hs, "offline") + t.deepEqual( + hs.queue.toArray(), + [["Discord", first], ["Discord", second]], + "still at the front, and not duplicated", + ) + + const replayed = [] + stubBridge(packet => replayed.push(packet)) + await replayQueue(t, hs) + t.deepEqual(replayed, [first, second]) + restore() +}) diff --git a/test/test.js b/test/test.js index 96326f5..268b603 100644 --- a/test/test.js +++ b/test/test.js @@ -140,6 +140,7 @@ file._actuallyUploadDiscordFileToMxc = function(url, res) { throw new Error(`Not require("../src/matrix/kstate.test") require("../src/matrix/api.test") require("../src/matrix/file.test") + require("../src/matrix/homeserver-status.test") require("../src/matrix/mreq.test") require("../src/matrix/read-registration.test") require("../src/matrix/room-upgrade.test")