From caf5f4ad7149ead0b646f41b01ae5fe1b9cd7759 Mon Sep 17 00:00:00 2001 From: l1s5 Date: Tue, 27 Oct 2026 15:10:25 +0900 Subject: [PATCH 1/6] matrix: fix homeserver checks `checkingWhileOnline` compared against "check while online", but the transition is called "check assume online", so it was always false. --- src/matrix/homeserver-status.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/matrix/homeserver-status.js b/src/matrix/homeserver-status.js index c50d5ef..5769679 100644 --- a/src/matrix/homeserver-status.js +++ b/src/matrix/homeserver-status.js @@ -39,7 +39,7 @@ 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) { -- 2.34.1 From a723c1c1a65b3a4689de037f82b55b0aedfa2c5b Mon Sep 17 00:00:00 2001 From: l1s5 Date: Tue, 27 Oct 2026 16:46:22 +0900 Subject: [PATCH 2/6] matrix: queue d2m events while checking the homeserver During a "check assume offline" at startup, `isRealTime()` is false, so incoming Discord events get queued, but `queuePacket()` asserted that we were offline or recovering. The updated assertion implicitly adds `checking` to the list of valid states, allowing packets to be queued in this state. --- src/matrix/homeserver-status.js | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/matrix/homeserver-status.js b/src/matrix/homeserver-status.js index 5769679..e234067 100644 --- a/src/matrix/homeserver-status.js +++ b/src/matrix/homeserver-status.js @@ -42,7 +42,9 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { this.checkingWhileOnline = event === "check assume online" const pingResult = await api.ping().catch(e => ({ok: false})) this.lastCheck = Date.now() - if (pingResult.ok) { + 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") @@ -99,6 +101,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,10 +140,10 @@ 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. + * When not real time, call this for incoming packets to queue them to be sent in order later. */ queuePacket(sourceAndPacket) { - assert(["offline", "recovering"].includes(this.sm.currentStateName)) + assert(!this.isRealTime()) this.queue.push(sourceAndPacket) } -- 2.34.1 From 2c650742f0cbe892b13971989dd7ac8be4302180 Mon Sep 17 00:00:00 2001 From: l1s5 Date: Wed, 28 Oct 2026 07:48:04 +0900 Subject: [PATCH 3/6] matrix: refactor d2m packet enqueuing Merge `setErrorWithPacket` into `queuePacket`. Also set a more accurate type. --- src/m2d/event-dispatcher.js | 2 +- src/matrix/homeserver-status.js | 27 ++++++++++++++++----------- 2 files changed, 17 insertions(+), 12 deletions(-) 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 e234067..69bd431 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 @@ -140,19 +140,24 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { } /** - * When not real time, 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(!this.isRealTime()) - 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) + 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) } }) -- 2.34.1 From e1f14ff1b9fa33d43f7848762bb745812be6478c Mon Sep 17 00:00:00 2001 From: l1s5 Date: Wed, 28 Oct 2026 08:18:33 +0900 Subject: [PATCH 4/6] matrix: ignore stale ping results in homeserver status machinery On the off chance that something invokes `waitForOnline` during `checking` or `offline` (e.g. on a `GUILD_CREATE` event from discord), the state might have changed in between we send a ping and receive it. This commit adds guards to not act on such a stale ping result. --- src/matrix/homeserver-status.js | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/matrix/homeserver-status.js b/src/matrix/homeserver-status.js index 69bd431..163a1a0 100644 --- a/src/matrix/homeserver-status.js +++ b/src/matrix/homeserver-status.js @@ -42,6 +42,8 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { this.checkingWhileOnline = event === "check assume online" const pingResult = await api.ping().catch(e => ({ok: false})) this.lastCheck = Date.now() + // 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) { @@ -60,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") } -- 2.34.1 From 01da1ae497ed4c78b26c562ae47a18c208c611cf Mon Sep 17 00:00:00 2001 From: l1s5 Date: Thu, 29 Oct 2026 21:32:26 +0900 Subject: [PATCH 5/6] matrix: await queued m2d packets This ensures that all m2d packets in the queue arrives in the order they were queued, as well as being able to catch and log any errors. --- src/matrix/homeserver-status.js | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/matrix/homeserver-status.js b/src/matrix/homeserver-status.js index 163a1a0..f29b580 100644 --- a/src/matrix/homeserver-status.js +++ b/src/matrix/homeserver-status.js @@ -82,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() -- 2.34.1 From a04daf2fb871f349a9c47885e35c07cc5e84b801 Mon Sep 17 00:00:00 2001 From: l1s5 Date: Thu, 29 Oct 2026 23:30:03 +0900 Subject: [PATCH 6/6] matrix: write tests for homeserver status machinery --- src/matrix/homeserver-status.js | 7 +- src/matrix/homeserver-status.test.js | 247 +++++++++++++++++++++++++++ test/test.js | 1 + 3 files changed, 253 insertions(+), 2 deletions(-) create mode 100644 src/matrix/homeserver-status.test.js diff --git a/src/matrix/homeserver-status.js b/src/matrix/homeserver-status.js index f29b580..1f6db93 100644 --- a/src/matrix/homeserver-status.js +++ b/src/matrix/homeserver-status.js @@ -19,7 +19,7 @@ function debugHomeserverStatus(message) { } } -const homeserverStatus = sync.remember(() => new class HomeserverStatus { +class HomeserverStatus { constructor() { /** @private */ this.queue = new Denque() @@ -168,6 +168,9 @@ const homeserverStatus = sync.remember(() => new class HomeserverStatus { this.queue.push(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") -- 2.34.1