Various homeserver state FST and queue fixes #101
2 changed files with 16 additions and 11 deletions
matrix: refactor d2m packet enqueuing
Merge `setErrorWithPacket` into `queuePacket`. Also set a more accurate type.
commit
2c650742f0
|
|
@ -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")
|
|| (e instanceof mreq.MatrixServerError && [502, 503].includes(e.httpStatus) && e.errcode === "CX_SERVER_ERROR")
|
||||||
) {
|
) {
|
||||||
// Matrix homeserver is down (reverse proxy indicated failure)
|
// Matrix homeserver is down (reverse proxy indicated failure)
|
||||||
homeserverStatus.homeserverStatus.setErrorWithPacket([source, payload])
|
homeserverStatus.homeserverStatus.queuePacket([source, payload])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,8 @@
|
||||||
// @ts-check
|
// @ts-check
|
||||||
|
|
||||||
const assert = require("assert").strict
|
|
||||||
const Denque = require("denque")
|
const Denque = require("denque")
|
||||||
const StateMachine = require("snowtransfer").StateMachine
|
const StateMachine = require("snowtransfer").StateMachine
|
||||||
|
const Ty = require("../types")
|
||||||
|
|
||||||
const passthrough = require("../passthrough")
|
const passthrough = require("../passthrough")
|
||||||
const {sync} = 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<any>]} sourceAndPacket
|
||||||
*/
|
*/
|
||||||
queuePacket(sourceAndPacket) {
|
queuePacket(sourceAndPacket) {
|
||||||
assert(!this.isRealTime())
|
if (this.sm.currentStateName === "recovering" && this.queue.peekFront()?.[1] === sourceAndPacket[1]) {
|
||||||
this.queue.push(sourceAndPacket)
|
this.sm.doTransition("error")
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
setErrorWithPacket(sourceAndPacket) {
|
if (this.isRealTime()) {
|
||||||
const wasRecovering = this.sm.currentStateName === "recovering"
|
|
||||||
this.sm.doTransition("error")
|
this.sm.doTransition("error")
|
||||||
if (!wasRecovering) { // if was recovering then sourceAndPacket is already in the right place in queue
|
|
||||||
this.queuePacket(sourceAndPacket)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
this.queue.push(sourceAndPacket)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue