mirrors/signal-desktop
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

Wrap eventHandlerQueue jobs with task with timeout

Browse source
This commit is contained in:
Fedor Indutny 2022-11-09 10:59:32 -08:00 • committed by GitHub
parent 5cee260063
commit d7a2669b49
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 112 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

97
ts/background.ts
View file

@ -291,23 +291,25 @@ export async function startApp(): Promise<void> {
serverTrustRoot: window.getServerTrustRoot(),
});
function queuedEventListener<Args extends Array<unknown>>(
handler: (...args: Args) => Promise<void> | void,
function queuedEventListener<E extends Event>(
handler: (event: E) => Promise<void> | void,
track = true
): (...args: Args) => void {
return (...args: Args): void => {
eventHandlerQueue.add(async () => {
try {
await handler(...args);
} finally {
// message/sent: Message.handleDataMessage has its own queue and will
// trigger this event itself when complete.
// error: Error processing (below) also has its own queue and self-trigger.
if (track) {
window.Whisper.events.trigger('incrementProgress');
): (event: E) => void {
return (event: E): void => {
eventHandlerQueue.add(
createTaskWithTimeout(async () => {
try {
await handler(event);
} finally {
// message/sent: Message.handleDataMessage has its own queue and will
// trigger this event itself when complete.
// error: Error processing (below) also has its own queue and self-trigger.
if (track) {
window.Whisper.events.trigger('incrementProgress');
}
}
}
});
}, `queuedEventListener(${event.type}, ${event.timeStamp})`)
);
};
}
@ -361,13 +363,13 @@ export async function startApp(): Promise<void> {
);
messageReceiver.addEventListener(
'decryption-error',
queuedEventListener((event: DecryptionErrorEvent) => {
queuedEventListener((event: DecryptionErrorEvent): void => {
onDecryptionErrorQueue.add(() => onDecryptionError(event));
})
);
messageReceiver.addEventListener(
'retry-request',
queuedEventListener((event: RetryRequestEvent) => {
queuedEventListener((event: RetryRequestEvent): void => {
onRetryRequestQueue.add(() => onRetryRequest(event));
})
);
@ -437,7 +439,6 @@ export async function startApp(): Promise<void> {
const eventHandlerQueue = new PQueue({
concurrency: 1,
timeout: durations.MINUTE * 30,
});
// Note: this queue is meant to allow for stop/start of tasks, not limit parallelism.
@ -2449,7 +2450,7 @@ export async function startApp(): Promise<void> {
window.waitForEmptyEventQueue = waitForEmptyEventQueue;
async function onEmpty() {
async function onEmpty(): Promise<void> {
const { storage } = window.textsecure;
await Promise.all([
@ -2586,7 +2587,7 @@ export async function startApp(): Promise<void> {
connect();
}
function onConfiguration(ev: ConfigurationEvent) {
function onConfiguration(ev: ConfigurationEvent): void {
ev.confirm();
const { configuration } = ev;
@ -2618,7 +2619,7 @@ export async function startApp(): Promise<void> {
}
}
function onTyping(ev: TypingEvent) {
function onTyping(ev: TypingEvent): void {
// Note: this type of message is automatically removed from cache in MessageReceiver
const { typing, sender, senderUuid, senderDevice } = ev;
@ -2707,7 +2708,7 @@ export async function startApp(): Promise<void> {
});
}
async function onStickerPack(ev: StickerPackEvent) {
function onStickerPack(ev: StickerPackEvent): void {
ev.confirm();
const packs = ev.stickerPacks;
@ -2741,13 +2742,13 @@ export async function startApp(): Promise<void> {
});
}
async function onGroupSyncComplete() {
async function onGroupSyncComplete(): Promise<void> {
log.info('onGroupSyncComplete');
await window.storage.put('synced_at', Date.now());
}
// Note: this handler is only for v1 groups received via 'group sync' messages
async function onGroupReceived(ev: GroupEvent) {
async function onGroupReceived(ev: GroupEvent): Promise<void> {
const details = ev.groupDetails;
const { id } = details;
@ -2868,7 +2869,7 @@ export async function startApp(): Promise<void> {
maxSize: Infinity,
});
function onEnvelopeReceived({ envelope }: EnvelopeEvent) {
function onEnvelopeReceived({ envelope }: EnvelopeEvent): void {
const ourUuid = window.textsecure.storage.user.getUuid()?.toString();
if (envelope.sourceUuid && envelope.sourceUuid !== ourUuid) {
window.ConversationController.maybeMergeContacts({
@ -2882,7 +2883,7 @@ export async function startApp(): Promise<void> {
// Note: We do very little in this function, since everything in handleDataMessage is
// inside a conversation-specific queue(). Any code here might run before an earlier
// message is processed in handleDataMessage().
function onMessageReceived(event: MessageEvent) {
async function onMessageReceived(event: MessageEvent): Promise<void> {
const { data, confirm } = event;
const messageDescriptor = getMessageDescriptor({
@ -2947,7 +2948,7 @@ export async function startApp(): Promise<void> {
if (!isValidReactionEmoji(reaction.emoji)) {
log.warn('Received an invalid reaction emoji. Dropping it');
confirm();
return Promise.resolve();
return;
}
strictAssert(
@ -2975,7 +2976,7 @@ export async function startApp(): Promise<void> {
// Note: We do not wait for completion here
Reactions.getSingleton().onReaction(reactionModel, message);
confirm();
return Promise.resolve();
return;
}
if (data.message.delete) {
@ -3004,21 +3005,22 @@ export async function startApp(): Promise<void> {
Deletes.getSingleton().onDelete(deleteModel);
confirm();
return Promise.resolve();
return;
}
if (handleGroupCallUpdateMessage(data.message, messageDescriptor)) {
confirm();
return Promise.resolve();
return;
}
// Don't wait for handleDataMessage, as it has its own per-conversation queueing
message.handleDataMessage(data.message, event.confirm);
return Promise.resolve();
}
async function onProfileKeyUpdate({ data, confirm }: ProfileKeyUpdateEvent) {
async function onProfileKeyUpdate({
data,
confirm,
}: ProfileKeyUpdateEvent): Promise<void> {
const conversation = window.ConversationController.maybeMergeContacts({
aci: data.sourceUuid,
e164: data.source,
@ -3279,7 +3281,7 @@ export async function startApp(): Promise<void> {
// Note: We do very little in this function, since everything in handleDataMessage is
// inside a conversation-specific queue(). Any code here might run before an earlier
// message is processed in handleDataMessage().
function onSentMessage(event: SentEvent) {
async function onSentMessage(event: SentEvent): Promise<void> {
const { data, confirm } = event;
const source = window.textsecure.storage.user.getNumber();
@ -3327,7 +3329,7 @@ export async function startApp(): Promise<void> {
if (!isValidReactionEmoji(reaction.emoji)) {
log.warn('Received an invalid reaction emoji. Dropping it');
event.confirm();
return Promise.resolve();
return;
}
log.info('Queuing sent reaction for', reaction.targetTimestamp);
@ -3345,7 +3347,7 @@ export async function startApp(): Promise<void> {
Reactions.getSingleton().onReaction(reactionModel, message);
event.confirm();
return Promise.resolve();
return;
}
if (data.message.delete) {
@ -3367,20 +3369,18 @@ export async function startApp(): Promise<void> {
// Note: We do not wait for completion here
Deletes.getSingleton().onDelete(deleteModel);
confirm();
return Promise.resolve();
return;
}
if (handleGroupCallUpdateMessage(data.message, messageDescriptor)) {
event.confirm();
return Promise.resolve();
return;
}
// Don't wait for handleDataMessage, as it has its own per-conversation queueing
message.handleDataMessage(data.message, event.confirm, {
data,
});
return Promise.resolve();
}
type MessageDescriptor = {
@ -3525,7 +3525,7 @@ export async function startApp(): Promise<void> {
}
}
function onError(ev: ErrorEvent) {
function onError(ev: ErrorEvent): void {
const { error } = ev;
log.error('background onError:', Errors.toLogFormat(error));
@ -3540,7 +3540,7 @@ export async function startApp(): Promise<void> {
log.warn('background onError: Doing nothing with incoming error');
}
async function onViewOnceOpenSync(ev: ViewOnceOpenSyncEvent) {
function onViewOnceOpenSync(ev: ViewOnceOpenSyncEvent): void {
ev.confirm();
const { source, sourceUuid, timestamp } = ev;
@ -3558,7 +3558,7 @@ export async function startApp(): Promise<void> {
ViewOnceOpenSyncs.getSingleton().onSync(sync);
}
async function onFetchLatestSync(ev: FetchLatestEvent) {
async function onFetchLatestSync(ev: FetchLatestEvent): Promise<void> {
ev.confirm();
const { eventType } = ev;
@ -3567,6 +3567,7 @@ export async function startApp(): Promise<void> {
switch (eventType) {
case FETCH_LATEST_ENUM.LOCAL_PROFILE: {
log.info('onFetchLatestSync: fetching latest local profile');
const ourUuid = window.textsecure.storage.user.getUuid()?.toString();
const ourE164 = window.textsecure.storage.user.getNumber();
await Promise.all([
@ -3620,7 +3621,7 @@ export async function startApp(): Promise<void> {
}
}
async function onMessageRequestResponse(ev: MessageRequestResponseEvent) {
function onMessageRequestResponse(ev: MessageRequestResponseEvent): void {
ev.confirm();
const {
@ -3656,7 +3657,7 @@ export async function startApp(): Promise<void> {
MessageRequests.getSingleton().onResponse(sync);
}
function onReadReceipt(event: Readonly<ReadEvent>) {
function onReadReceipt(event: Readonly<ReadEvent>): void {
onReadOrViewReceipt({
logTitle: 'read receipt',
event,
@ -3731,7 +3732,7 @@ export async function startApp(): Promise<void> {
MessageReceipts.getSingleton().onReceipt(receipt);
}
function onReadSync(ev: ReadSyncEvent) {
function onReadSync(ev: ReadSyncEvent): Promise<void> {
const { envelopeTimestamp, sender, senderUuid, timestamp } = ev.read;
const readAt = envelopeTimestamp;
const senderConversation = window.ConversationController.lookupOrCreate({
@ -3770,7 +3771,7 @@ export async function startApp(): Promise<void> {
return ReadSyncs.getSingleton().onSync(receipt);
}
function onViewSync(ev: ViewSyncEvent) {
function onViewSync(ev: ViewSyncEvent): Promise<void> {
const { envelopeTimestamp, senderE164, senderUuid, timestamp } = ev.view;
const senderConversation = window.ConversationController.lookupOrCreate({
e164: senderE164,
@ -3808,7 +3809,7 @@ export async function startApp(): Promise<void> {
return ViewSyncs.getSingleton().onSync(receipt);
}
function onDeliveryReceipt(ev: DeliveryEvent) {
function onDeliveryReceipt(ev: DeliveryEvent): void {
const { deliveryReceipt } = ev;
const {
envelopeTimestamp,

11
ts/models/conversations.ts
View file

@ -3498,10 +3498,9 @@ export class ConversationModel extends window.Backbone
): Promise<T> {
this.jobQueue = this.jobQueue || new PQueue({ concurrency: 1 });
const taskWithTimeout = createTaskWithTimeout(
callback,
`conversation ${this.idForLogging()}`
);
const logId = `conversation.queueJob(${this.idForLogging()}, ${name})`;
const taskWithTimeout = createTaskWithTimeout(callback, logId);
const abortController = new AbortController();
const { signal: abortSignal } = abortController;
@ -3512,7 +3511,7 @@ export class ConversationModel extends window.Backbone
const waitTime = startedAt - queuedAt;
if (waitTime > JOB_REPORTING_THRESHOLD_MS) {
log.info(`Conversation job ${name} was blocked for ${waitTime}ms`);
log.info(`${logId}: was blocked for ${waitTime}ms`);
}
try {
@ -3524,7 +3523,7 @@ export class ConversationModel extends window.Backbone
const duration = Date.now() - startedAt;
if (duration > JOB_REPORTING_THRESHOLD_MS) {
log.info(`Conversation job ${name} took ${duration}ms`);
log.info(`${logId}: took ${duration}ms`);
}
}
});

2
ts/models/messages.ts
View file

@ -2203,7 +2203,7 @@ export class MessageModel extends window.Backbone.Model<MessageAttributesType> {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const conversation = window.ConversationController.get(conversationId)!;
const idLog = `handleDataMessage/${conversation.idForLogging()} ${message.idForLogging()}`;
await conversation.queueJob('handleDataMessage', async () => {
await conversation.queueJob(idLog, async () => {
log.info(`${idLog}: starting processing in queue`);
// First, check for duplicates. If we find one, stop processing here.

2
ts/services/contactSync.ts
View file

@ -187,6 +187,8 @@ async function doContactSync({
await window.storage.put('synced_at', Date.now());
window.Whisper.events.trigger('contactSync:complete');
log.info(`${logId}: done`);
}
export async function onContactSync(ev: ContactSyncEvent): Promise<void> {

104
ts/test-both/processDataMessage_test.ts
View file

@ -46,16 +46,16 @@ describe('processDataMessage', () => {
TIMESTAMP
);
it('should process attachments', async () => {
const out = await check({
it('should process attachments', () => {
const out = check({
attachments: [UNPROCESSED_ATTACHMENT],
});
assert.deepStrictEqual(out.attachments, [PROCESSED_ATTACHMENT]);
});
it('should process attachments with 0 cdnId', async () => {
const out = await check({
it('should process attachments with 0 cdnId', () => {
const out = check({
attachments: [
{
...UNPROCESSED_ATTACHMENT,
@ -72,25 +72,24 @@ describe('processDataMessage', () => {
]);
});
it('should throw on too many attachments', async () => {
it('should throw on too many attachments', () => {
const attachments: Array<Proto.IAttachmentPointer> = [];
for (let i = 0; i < ATTACHMENT_MAX + 1; i += 1) {
attachments.push(UNPROCESSED_ATTACHMENT);
}
await assert.isRejected(
check({ attachments }),
assert.throws(
() => check({ attachments }),
`Too many attachments: ${ATTACHMENT_MAX + 1} included in one message` +
`, max is ${ATTACHMENT_MAX}`
);
});
it('should process group context UPDATE/QUIT message', async () => {
it('should process group context UPDATE/QUIT message', () => {
const { UPDATE, QUIT } = Proto.GroupContext.Type;
for (const type of [UPDATE, QUIT]) {
// eslint-disable-next-line no-await-in-loop
const out = await check({
const out = check({
body: 'should be deleted',
attachments: [UNPROCESSED_ATTACHMENT],
group: {
@ -115,8 +114,8 @@ describe('processDataMessage', () => {
}
});
it('should process group context DELIVER message', async () => {
const out = await check({
it('should process group context DELIVER message', () => {
const out = check({
body: 'should not be deleted',
attachments: [UNPROCESSED_ATTACHMENT],
group: {
@ -139,8 +138,8 @@ describe('processDataMessage', () => {
});
});
it('should process groupv2 context', async () => {
const out = await check({
it('should process groupv2 context', () => {
const out = check({
groupV2: {
masterKey: new Uint8Array(32),
revision: 1,
@ -168,16 +167,16 @@ describe('processDataMessage', () => {
});
});
it('should base64 profileKey', async () => {
const out = await check({
it('should base64 profileKey', () => {
const out = check({
profileKey: new Uint8Array([42, 23, 55]),
});
assert.strictEqual(out.profileKey, 'Khc3');
});
it('should process quote', async () => {
const out = await check({
it('should process quote', () => {
const out = check({
quote: {
id: Long.fromNumber(1),
authorUuid: 'author',
@ -208,8 +207,8 @@ describe('processDataMessage', () => {
});
});
it('should process contact', async () => {
const out = await check({
it('should process contact', () => {
const out = check({
contact: [
{
avatar: {
@ -235,16 +234,14 @@ describe('processDataMessage', () => {
]);
});
it('should process reaction', async () => {
it('should process reaction', () => {
assert.deepStrictEqual(
(
await check({
reaction: {
emoji: '😎',
targetTimestamp: Long.fromNumber(TIMESTAMP),
},
})
).reaction,
check({
reaction: {
emoji: '😎',
targetTimestamp: Long.fromNumber(TIMESTAMP),
},
}).reaction,
{
emoji: '😎',
remove: false,
@ -254,15 +251,13 @@ describe('processDataMessage', () => {
);
assert.deepStrictEqual(
(
await check({
reaction: {
emoji: '😎',
remove: true,
targetTimestamp: Long.fromNumber(TIMESTAMP),
},
})
).reaction,
check({
reaction: {
emoji: '😎',
remove: true,
targetTimestamp: Long.fromNumber(TIMESTAMP),
},
}).reaction,
{
emoji: '😎',
remove: true,
@ -272,8 +267,8 @@ describe('processDataMessage', () => {
);
});
it('should process preview', async () => {
const out = await check({
it('should process preview', () => {
const out = check({
preview: [
{
date: Long.fromNumber(TIMESTAMP),
@ -293,8 +288,8 @@ describe('processDataMessage', () => {
]);
});
it('should process sticker', async () => {
const out = await check({
it('should process sticker', () => {
const out = check({
sticker: {
packId: new Uint8Array([1, 2, 3]),
packKey: new Uint8Array([4, 5, 6]),
@ -313,8 +308,8 @@ describe('processDataMessage', () => {
});
});
it('should process FLAGS=END_SESSION', async () => {
const out = await check({
it('should process FLAGS=END_SESSION', () => {
const out = check({
flags: FLAGS.END_SESSION,
body: 'should be deleted',
group: {
@ -329,11 +324,10 @@ describe('processDataMessage', () => {
assert.deepStrictEqual(out.attachments, []);
});
it('should process FLAGS=EXPIRATION_TIMER_UPDATE,PROFILE_KEY_UPDATE', async () => {
it('should process FLAGS=EXPIRATION_TIMER_UPDATE,PROFILE_KEY_UPDATE', () => {
const values = [FLAGS.EXPIRATION_TIMER_UPDATE, FLAGS.PROFILE_KEY_UPDATE];
for (const flags of values) {
// eslint-disable-next-line no-await-in-loop
const out = await check({
const out = check({
flags,
body: 'should be deleted',
attachments: [UNPROCESSED_ATTACHMENT],
@ -344,15 +338,15 @@ describe('processDataMessage', () => {
}
});
it('processes trivial fields', async () => {
assert.strictEqual((await check({ flags: null })).flags, 0);
assert.strictEqual((await check({ flags: 1 })).flags, 1);
it('processes trivial fields', () => {
assert.strictEqual(check({ flags: null }).flags, 0);
assert.strictEqual(check({ flags: 1 }).flags, 1);
assert.strictEqual((await check({ expireTimer: null })).expireTimer, 0);
assert.strictEqual((await check({ expireTimer: 123 })).expireTimer, 123);
assert.strictEqual(check({ expireTimer: null }).expireTimer, 0);
assert.strictEqual(check({ expireTimer: 123 }).expireTimer, 123);
assert.isFalse((await check({ isViewOnce: null })).isViewOnce);
assert.isFalse((await check({ isViewOnce: false })).isViewOnce);
assert.isTrue((await check({ isViewOnce: true })).isViewOnce);
assert.isFalse(check({ isViewOnce: null }).isViewOnce);
assert.isFalse(check({ isViewOnce: false }).isViewOnce);
assert.isTrue(check({ isViewOnce: true }).isViewOnce);
});
});

8
ts/textsecure/MessageReceiver.ts
View file

@ -1935,7 +1935,7 @@ export default class MessageReceiver
}
await p;
const message = await this.processDecrypted(envelope, msg);
const message = this.processDecrypted(envelope, msg);
const groupId = this.getProcessedGroupId(message);
const isBlocked = groupId ? this.isGroupBlocked(groupId) : false;
const { source, sourceUuid } = envelope;
@ -2234,7 +2234,7 @@ export default class MessageReceiver
logUnexpectedUrgentValue(envelope, type);
const message = await this.processDecrypted(envelope, msg);
const message = this.processDecrypted(envelope, msg);
const groupId = this.getProcessedGroupId(message);
const isBlocked = groupId ? this.isGroupBlocked(groupId) : false;
const { source, sourceUuid } = envelope;
@ -3324,10 +3324,10 @@ export default class MessageReceiver
await this.storage.protocol.archiveAllSessions(theirUuid);
}
private async processDecrypted(
private processDecrypted(
envelope: ProcessedEnvelope,
decrypted: Proto.IDataMessage
): Promise<ProcessedDataMessage> {
): ProcessedDataMessage {
return processDataMessage(decrypted, envelope.timestamp);
}
}

4
ts/textsecure/processDataMessage.ts
View file

@ -267,10 +267,10 @@ export function processGiftBadge(
};
}
export async function processDataMessage(
export function processDataMessage(
message: Proto.IDataMessage,
envelopeTimestamp: number
): Promise<ProcessedDataMessage> {
): ProcessedDataMessage {
/* eslint-disable no-bitwise */
// Now that its decrypted, validate the message and clean it up for consumer