From 1f45bce0a2e75bd273bcba32334ccb2b0cda6b70 Mon Sep 17 00:00:00 2001 From: Evan Hahn <69474926+EvanHahn-Signal@users.noreply.github.com> Date: Thu, 2 Sep 2021 17:31:21 -0500 Subject: [PATCH] When a job fails, respect the `Retry-After` header if applicable --- ts/challenge.ts | 4 +- ts/jobs/helpers/commonShouldJobContinue.ts | 9 +- .../helpers/handleCommonJobRequestError.ts | 16 ++- ts/jobs/helpers/readAndViewSyncHelpers.ts | 7 +- .../sleepFor413RetryAfterTimeIfApplicable.ts | 37 +++++ ts/jobs/normalMessageSendJobQueue.ts | 17 ++- ts/jobs/viewedReceiptsJobQueue.ts | 7 +- ts/test-both/util/parseRetryAfter_test.ts | 11 +- ...epFor413RetryAfterTimeIfApplicable_test.ts | 133 ++++++++++++++++++ ts/textsecure/Errors.ts | 9 +- ts/textsecure/WebAPI.ts | 2 +- ts/util/parseRetryAfter.ts | 6 +- 12 files changed, 228 insertions(+), 30 deletions(-) create mode 100644 ts/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable.ts create mode 100644 ts/test-node/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable_test.ts diff --git a/ts/challenge.ts b/ts/challenge.ts index d3fb54b446e6..0b7971aecf18 100644 --- a/ts/challenge.ts +++ b/ts/challenge.ts @@ -468,9 +468,7 @@ export class ChallengeHandler { throw error; } - const retryAfter = parseRetryAfter( - error.responseHeaders['retry-after'].toString() - ); + const retryAfter = parseRetryAfter(error.responseHeaders['retry-after']); window.log.info(`challenge: retry after ${retryAfter}ms`); this.options.onChallengeFailed(retryAfter); diff --git a/ts/jobs/helpers/commonShouldJobContinue.ts b/ts/jobs/helpers/commonShouldJobContinue.ts index 3dd5dd9a0492..f16000544c09 100644 --- a/ts/jobs/helpers/commonShouldJobContinue.ts +++ b/ts/jobs/helpers/commonShouldJobContinue.ts @@ -10,17 +10,12 @@ import { isDone as isDeviceLinked } from '../../util/registration'; export async function commonShouldJobContinue({ attempt, log, - maxRetryTime, - timestamp, + timeRemaining, }: Readonly<{ attempt: number; log: LoggerType; - maxRetryTime: number; - timestamp: number; + timeRemaining: number; }>): Promise { - const maxJobAge = timestamp + maxRetryTime; - const timeRemaining = maxJobAge - Date.now(); - if (timeRemaining <= 0) { log.info("giving up because it's been too long"); return false; diff --git a/ts/jobs/helpers/handleCommonJobRequestError.ts b/ts/jobs/helpers/handleCommonJobRequestError.ts index ba805c03a771..7a0646653bb6 100644 --- a/ts/jobs/helpers/handleCommonJobRequestError.ts +++ b/ts/jobs/helpers/handleCommonJobRequestError.ts @@ -3,11 +3,17 @@ import type { LoggerType } from '../../logging/log'; import { parseIntWithFallback } from '../../util/parseIntWithFallback'; +import { sleepFor413RetryAfterTimeIfApplicable } from './sleepFor413RetryAfterTimeIfApplicable'; -export function handleCommonJobRequestError( - err: unknown, - log: LoggerType -): void { +export async function handleCommonJobRequestError({ + err, + log, + timeRemaining, +}: Readonly<{ + err: unknown; + log: LoggerType; + timeRemaining: number; +}>): Promise { if (!(err instanceof Error)) { throw err; } @@ -18,5 +24,7 @@ export function handleCommonJobRequestError( return; } + await sleepFor413RetryAfterTimeIfApplicable({ err, log, timeRemaining }); + throw err; } diff --git a/ts/jobs/helpers/readAndViewSyncHelpers.ts b/ts/jobs/helpers/readAndViewSyncHelpers.ts index 87b895aa2ba4..3361b64377dd 100644 --- a/ts/jobs/helpers/readAndViewSyncHelpers.ts +++ b/ts/jobs/helpers/readAndViewSyncHelpers.ts @@ -90,11 +90,12 @@ export async function runReadOrViewSyncJob({ return; } + const timeRemaining = timestamp + maxRetryTime - Date.now(); + const shouldContinue = await commonShouldJobContinue({ attempt, log, - maxRetryTime, - timestamp, + timeRemaining, }); if (!shouldContinue) { return; @@ -117,6 +118,6 @@ export async function runReadOrViewSyncJob({ }) ); } catch (err: unknown) { - handleCommonJobRequestError(err, log); + await handleCommonJobRequestError({ err, log, timeRemaining }); } } diff --git a/ts/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable.ts b/ts/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable.ts new file mode 100644 index 000000000000..4d9c2c2204f8 --- /dev/null +++ b/ts/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable.ts @@ -0,0 +1,37 @@ +// Copyright 2021 Signal Messenger, LLC +// SPDX-License-Identifier: AGPL-3.0-only + +import type { LoggerType } from '../../logging/log'; +import { sleep } from '../../util/sleep'; +import { parseRetryAfter } from '../../util/parseRetryAfter'; +import { isRecord } from '../../util/isRecord'; + +export async function sleepFor413RetryAfterTimeIfApplicable({ + err, + log, + timeRemaining, +}: Readonly<{ + err: unknown; + log: Pick; + timeRemaining: number; +}>): Promise { + if ( + timeRemaining <= 0 || + !(err instanceof Error) || + err.code !== 413 || + !isRecord(err.responseHeaders) + ) { + return; + } + + const retryAfter = Math.min( + parseRetryAfter(err.responseHeaders['retry-after']), + timeRemaining + ); + + log.info( + `Got a 413 response code. Sleeping for ${retryAfter} millisecond(s)` + ); + + await sleep(retryAfter); +} diff --git a/ts/jobs/normalMessageSendJobQueue.ts b/ts/jobs/normalMessageSendJobQueue.ts index 8f2a1f8c5cb2..48881d137e3f 100644 --- a/ts/jobs/normalMessageSendJobQueue.ts +++ b/ts/jobs/normalMessageSendJobQueue.ts @@ -7,6 +7,7 @@ import PQueue from 'p-queue'; import type { LoggerType } from '../logging/log'; import { exponentialBackoffMaxAttempts } from '../util/exponentialBackoff'; import { commonShouldJobContinue } from './helpers/commonShouldJobContinue'; +import { sleepFor413RetryAfterTimeIfApplicable } from './helpers/sleepFor413RetryAfterTimeIfApplicable'; import type { MessageModel } from '../models/messages'; import { getMessageById } from '../messages/getMessageById'; import type { ConversationModel } from '../models/conversations'; @@ -123,6 +124,7 @@ export class NormalMessageSendJobQueue extends JobQueue { + const timeRemaining = timestamp + MAX_RETRY_TIME - Date.now(); const isFinalAttempt = attempt >= MAX_ATTEMPTS; // We don't immediately use this value because we may want to mark the message @@ -130,8 +132,7 @@ export class NormalMessageSendJobQueue extends JobQueue + messageSendError instanceof Error && messageSendError.code === 413 + ); + await sleepFor413RetryAfterTimeIfApplicable({ + err: maybe413Error, + log, + timeRemaining, + }); + } + throw err; } }); diff --git a/ts/jobs/viewedReceiptsJobQueue.ts b/ts/jobs/viewedReceiptsJobQueue.ts index 54442cfaff88..97740f336c11 100644 --- a/ts/jobs/viewedReceiptsJobQueue.ts +++ b/ts/jobs/viewedReceiptsJobQueue.ts @@ -39,11 +39,12 @@ export class ViewedReceiptsJobQueue extends JobQueue { }: Readonly<{ data: ViewedReceiptsJobData; timestamp: number }>, { attempt, log }: Readonly<{ attempt: number; log: LoggerType }> ): Promise { + const timeRemaining = timestamp + MAX_RETRY_TIME - Date.now(); + const shouldContinue = await commonShouldJobContinue({ attempt, log, - maxRetryTime: MAX_RETRY_TIME, - timestamp, + timeRemaining, }); if (!shouldContinue) { return; @@ -52,7 +53,7 @@ export class ViewedReceiptsJobQueue extends JobQueue { try { await sendViewedReceipt(data.viewedReceipt); } catch (err: unknown) { - handleCommonJobRequestError(err, log); + await handleCommonJobRequestError({ err, log, timeRemaining }); } } } diff --git a/ts/test-both/util/parseRetryAfter_test.ts b/ts/test-both/util/parseRetryAfter_test.ts index 855c003f5cf7..11bd59355d7e 100644 --- a/ts/test-both/util/parseRetryAfter_test.ts +++ b/ts/test-both/util/parseRetryAfter_test.ts @@ -6,16 +6,21 @@ import { assert } from 'chai'; import { parseRetryAfter } from '../../util/parseRetryAfter'; describe('parseRetryAfter', () => { - it('should return 0 on invalid input', () => { + it('should return 1 second when passed non-strings', () => { + assert.equal(parseRetryAfter(undefined), 1000); + assert.equal(parseRetryAfter(1234), 1000); + }); + + it('should return 1 second with invalid strings', () => { assert.equal(parseRetryAfter('nope'), 1000); assert.equal(parseRetryAfter('1ff'), 1000); }); - it('should return milleseconds on valid input', () => { + it('should return milliseconds on valid input', () => { assert.equal(parseRetryAfter('100'), 100000); }); - it('should return apply minimum value', () => { + it('should return 1 second at minimum', () => { assert.equal(parseRetryAfter('0'), 1000); assert.equal(parseRetryAfter('-1'), 1000); }); diff --git a/ts/test-node/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable_test.ts b/ts/test-node/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable_test.ts new file mode 100644 index 000000000000..e012587ab152 --- /dev/null +++ b/ts/test-node/jobs/helpers/sleepFor413RetryAfterTimeIfApplicable_test.ts @@ -0,0 +1,133 @@ +// Copyright 2021 Signal Messenger, LLC +// SPDX-License-Identifier: AGPL-3.0-only + +import { assert } from 'chai'; +import * as sinon from 'sinon'; +import { HTTPError } from '../../../textsecure/Errors'; +import * as durations from '../../../util/durations'; + +import { sleepFor413RetryAfterTimeIfApplicable } from '../../../jobs/helpers/sleepFor413RetryAfterTimeIfApplicable'; + +describe('sleepFor413RetryAfterTimeIfApplicable', () => { + const createLogger = () => ({ info: sinon.spy() }); + + let sandbox: sinon.SinonSandbox; + let clock: sinon.SinonFakeTimers; + + beforeEach(() => { + sandbox = sinon.createSandbox(); + clock = sandbox.useFakeTimers(); + }); + + afterEach(() => { + sandbox.restore(); + }); + + it('does nothing if not passed a 413 HTTP error', async () => { + const log = createLogger(); + + const errors = [ + undefined, + new Error('Normal error'), + new HTTPError('Uh oh', { code: 422, headers: {}, response: {} }), + ]; + await Promise.all( + errors.map(async err => { + await sleepFor413RetryAfterTimeIfApplicable({ + err, + log, + timeRemaining: 1234, + }); + }) + ); + + sinon.assert.notCalled(log.info); + }); + + it('waits for 1 second if receiving a 413 HTTP error without a Retry-After header', async () => { + const err = new HTTPError('Slow down', { + code: 413, + headers: {}, + response: {}, + }); + + let done = false; + + (async () => { + await sleepFor413RetryAfterTimeIfApplicable({ + err, + log: createLogger(), + timeRemaining: 1234, + }); + done = true; + })(); + + await clock.tickAsync(999); + assert.isFalse(done); + + await clock.tickAsync(2); + assert.isTrue(done); + }); + + it('waits for Retry-After seconds if receiving a 413', async () => { + const err = new HTTPError('Slow down', { + code: 413, + headers: { 'retry-after': '200' }, + response: {}, + }); + + let done = false; + + (async () => { + await sleepFor413RetryAfterTimeIfApplicable({ + err, + log: createLogger(), + timeRemaining: 123456789, + }); + done = true; + })(); + + await clock.tickAsync(199 * durations.SECOND); + assert.isFalse(done); + + await clock.tickAsync(2 * durations.SECOND); + assert.isTrue(done); + }); + + it("won't wait longer than the remaining time", async () => { + const err = new HTTPError('Slow down', { + code: 413, + headers: { 'retry-after': '99999' }, + response: {}, + }); + + let done = false; + + (async () => { + await sleepFor413RetryAfterTimeIfApplicable({ + err, + log: createLogger(), + timeRemaining: 3 * durations.SECOND, + }); + done = true; + })(); + + await clock.tickAsync(4 * durations.SECOND); + assert.isTrue(done); + }); + + it('logs how long it will wait', async () => { + const log = createLogger(); + const err = new HTTPError('Slow down', { + code: 413, + headers: { 'retry-after': '123' }, + response: {}, + }); + + sleepFor413RetryAfterTimeIfApplicable({ err, log, timeRemaining: 9999999 }); + await clock.nextAsync(); + + sinon.assert.calledOnce(log.info); + sinon.assert.calledWith(log.info, sinon.match(/123000 millisecond\(s\)/)); + }); +}); diff --git a/ts/textsecure/Errors.ts b/ts/textsecure/Errors.ts index 33a5cc892a95..a8b33a187ffa 100644 --- a/ts/textsecure/Errors.ts +++ b/ts/textsecure/Errors.ts @@ -6,7 +6,8 @@ import { parseRetryAfter } from '../util/parseRetryAfter'; -import { CallbackResultType } from './Types.d'; +import type { CallbackResultType } from './Types.d'; +import type { HeaderListType } from './WebAPI'; function appendStack(newError: Error, originalError: Error) { // eslint-disable-next-line no-param-reassign @@ -129,6 +130,8 @@ export class OutgoingMessageError extends ReplayableError { export class SendMessageNetworkError extends ReplayableError { identifier: string; + responseHeaders?: HeaderListType | undefined; + constructor(identifier: string, _m: unknown, httpError: Error) { super({ name: 'SendMessageNetworkError', @@ -137,6 +140,7 @@ export class SendMessageNetworkError extends ReplayableError { [this.identifier] = identifier.split('.'); this.code = httpError.code; + this.responseHeaders = httpError.responseHeaders; appendStack(this, httpError); } @@ -166,8 +170,7 @@ export class SendMessageChallengeError extends ReplayableError { const headers = httpError.responseHeaders || {}; - this.retryAfter = - Date.now() + parseRetryAfter((headers['retry-after'] ?? 0).toString()); + this.retryAfter = Date.now() + parseRetryAfter(headers['retry-after']); appendStack(this, httpError); } diff --git a/ts/textsecure/WebAPI.ts b/ts/textsecure/WebAPI.ts index 7bd4ca64609b..17c6f5ba84f2 100644 --- a/ts/textsecure/WebAPI.ts +++ b/ts/textsecure/WebAPI.ts @@ -286,7 +286,7 @@ function getContentType(response: Response) { } type FetchHeaderListType = { [name: string]: string }; -type HeaderListType = { [name: string]: string | ReadonlyArray }; +export type HeaderListType = { [name: string]: string | ReadonlyArray }; type HTTPCodeType = 'GET' | 'POST' | 'PUT' | 'DELETE' | 'PATCH'; type RedactUrl = (url: string) => string; diff --git a/ts/util/parseRetryAfter.ts b/ts/util/parseRetryAfter.ts index 862a395ab315..80cf6a767f91 100644 --- a/ts/util/parseRetryAfter.ts +++ b/ts/util/parseRetryAfter.ts @@ -6,7 +6,11 @@ import { isNormalNumber } from './isNormalNumber'; const ONE_SECOND = 1000; const MINIMAL_RETRY_AFTER = ONE_SECOND; -export function parseRetryAfter(value: string): number { +export function parseRetryAfter(value: unknown): number { + if (typeof value !== 'string') { + return MINIMAL_RETRY_AFTER; + } + let retryAfter = parseInt(value, 10); if (!isNormalNumber(retryAfter) || retryAfter.toString() !== value) { retryAfter = 0;