matrix: write tests for homeserver status machinery

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

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