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])
 		}
 	}
 }

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<any>]} 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

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<HomeserverStatus, "sm" | "queue"> & {
+ *   sm: HomeserverStatus["sm"],
+ *   queue: HomeserverStatus["queue"],
+ * }} ExposedHomeserverStatus
+ */
+
+/** @returns {{hs: ExposedHomeserverStatus, pings: PromiseWithResolvers<any>[]}} */
+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()
+})

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