mirrors/vector-im-hydrogen-web
1
0
Fork 0
mirror of https://github.com/vector-im/hydrogen-web.git synced 2024-12-23 11:35:04 +01:00
Code Issues Projects Releases Wiki Activity

prevent ignored signaling messages from blocking the queue

Browse Source 
signaling messages get ignored when they are not for the
currently active call id. In that case we currently don't
advance the lastProcessedSeqNr counter, as we had a problem
before where the counter would be brought out of sync with
seq numbers for other call ids.
However when we've previously processed a signalling message
(e.g. the counter is not undefined) and the first message in the queue
is to be ignored, it will prevent the subsequent messages from being
dequeued as their seq number is more than 1 away from the last
processed seq. This adds an additional counter for ignored seq numbers
that is also used to see if the next message is only 1 away from
the next seq value.

I am adding logging as well here to have a better overview in the future
This commit is contained in:
Bruno Windels 2023-02-07 14:13:49 +01:00
parent 072004a9c2
commit cadeae98bc
1 changed files with 37 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
src/matrix/calls/group/Member.ts
View File

@ -59,6 +59,9 @@ class MemberConnection {
public retryCount: number = 0; public retryCount: number = 0;
public peerCall?: PeerCall; public peerCall?: PeerCall;
public lastProcessedSeqNr: number | undefined; public lastProcessedSeqNr: number | undefined;
// XXX: Not needed anymore when seq is scoped to call_id
// see https://github.com/matrix-org/matrix-spec-proposals/pull/3401#discussion_r1097482166
public lastIgnoredSeqNr: number | undefined;
public queuedSignallingMessages: SignallingMessage<MGroupCallBase>[] = []; public queuedSignallingMessages: SignallingMessage<MGroupCallBase>[] = [];
public outboundSeqCounter: number = 0; public outboundSeqCounter: number = 0;
@ -73,16 +76,24 @@ class MemberConnection {
if (this.queuedSignallingMessages.length === 0) { if (this.queuedSignallingMessages.length === 0) {
return false; return false;
} }
const first = this.queuedSignallingMessages[0];
const firstSeq = first.content.seq;
// prevent not being able to jump over seq values of ignored messages for other call ids
// as they don't increase lastProcessedSeqNr.
if (this.lastIgnoredSeqNr !== undefined && firstSeq === this.lastIgnoredSeqNr + 1) {
return true;
}
if (this.lastProcessedSeqNr === undefined) { if (this.lastProcessedSeqNr === undefined) {
return true; return true;
} }
const first = this.queuedSignallingMessages[0];
// allow messages with both a seq we've just seen and // allow messages with both a seq we've just seen and
// the next one to be dequeued as it can happen // the next one to be dequeued as it can happen
// that messages for other callIds (which could repeat seq) // that messages for other callIds (which could repeat seq)
// are present in the queue // are present in the queue
return first.content.seq === this.lastProcessedSeqNr || // XXX: Not needed anymore when seq is scoped to call_id
first.content.seq === this.lastProcessedSeqNr + 1; // see https://github.com/matrix-org/matrix-spec-proposals/pull/3401#discussion_r1097482166
return firstSeq === this.lastProcessedSeqNr ||
firstSeq === this.lastProcessedSeqNr + 1;
} }
dispose() { dispose() {
@ -382,17 +393,29 @@ export class Member {
private dequeueSignallingMessages(connection: MemberConnection, peerCall: PeerCall, newMessage: SignallingMessage<MGroupCallBase>, syncLog: ILogItem): boolean { private dequeueSignallingMessages(connection: MemberConnection, peerCall: PeerCall, newMessage: SignallingMessage<MGroupCallBase>, syncLog: ILogItem): boolean {
let hasNewMessageBeenDequeued = false; let hasNewMessageBeenDequeued = false;
while (connection.canDequeueNextSignallingMessage) { while (connection.canDequeueNextSignallingMessage) {
const message = connection.queuedSignallingMessages.shift()!; syncLog.wrap("dequeue message", log => {
if (message === newMessage) { const message = connection.queuedSignallingMessages.shift()!;
hasNewMessageBeenDequeued = true; if (message === newMessage) {
} log.set("isNewMsg", true);
// ignore items in the queue that should not be handled and prevent hasNewMessageBeenDequeued = true;
// the lastProcessedSeqNr being corrupted with the `seq` for other call ids }
if (peerCall.getMessageAction(message) === IncomingMessageAction.Handle) { const seq = message.content?.seq;
const item = peerCall.handleIncomingSignallingMessage(message, this.deviceId, connection.logItem); log.set("seq", seq);
syncLog.refDetached(item); // ignore items in the queue that should not be handled and prevent
connection.lastProcessedSeqNr = message.content.seq; // the lastProcessedSeqNr being corrupted with the `seq` for other call ids
} // XXX: Not needed anymore when seq is scoped to call_id
// see https://github.com/matrix-org/matrix-spec-proposals/pull/3401#discussion_r1097482166
const action = peerCall.getMessageAction(message);
if (action === IncomingMessageAction.Handle) {
const item = peerCall.handleIncomingSignallingMessage(message, this.deviceId, connection.logItem);
log.refDetached(item);
connection.lastProcessedSeqNr = seq;
} else {
log.set("type", message.type);
log.set("ignored", true);
connection.lastIgnoredSeqNr = seq;
}
});
} }
return hasNewMessageBeenDequeued; return hasNewMessageBeenDequeued;
} }