From 3d8bce5ae468180f03098a2b9204ce63129d72ae Mon Sep 17 00:00:00 2001 From: zkldi <20380519+zkldi@users.noreply.github.com> Date: Tue, 1 Feb 2022 01:38:26 +0000 Subject: [PATCH 1/3] Imports while the user has an import ongoing should not be thrown away and should instead be queued properly. Fixes #600 --- .../framework/import-locks/lock.ts | 3 + .../score-import/framework/score-import.ts | 62 ++++++++++++++++--- .../score-importing/score-import-main.ts | 59 ++++++++++-------- 3 files changed, 89 insertions(+), 35 deletions(-) diff --git a/server/src/lib/score-import/framework/import-locks/lock.ts b/server/src/lib/score-import/framework/import-locks/lock.ts index be82069a5..8d054dbdd 100644 --- a/server/src/lib/score-import/framework/import-locks/lock.ts +++ b/server/src/lib/score-import/framework/import-locks/lock.ts @@ -1,4 +1,5 @@ import db from "external/mongo/db"; +import { rootLogger } from "lib/logger/logger"; import { integer } from "tachi-common"; /** @@ -31,6 +32,8 @@ export async function CheckAndSetOngoingImportLock(userID: integer) { } ); + rootLogger.crit("", lockWasSet); + return !lockWasSet; } diff --git a/server/src/lib/score-import/framework/score-import.ts b/server/src/lib/score-import/framework/score-import.ts index dd2533952..d687ee9f3 100644 --- a/server/src/lib/score-import/framework/score-import.ts +++ b/server/src/lib/score-import/framework/score-import.ts @@ -1,10 +1,15 @@ +/* eslint-disable no-await-in-loop */ import { ServerConfig } from "lib/setup/config"; import { ScoreImportJobData } from "../worker/types"; import { GetInputParser } from "./common/get-input-parser"; import ScoreImportMain from "./score-importing/score-import-main"; -import { ImportTypes, ImportDocument } from "tachi-common"; +import { ImportTypes, ImportDocument, integer } from "tachi-common"; import ScoreImportQueue, { ScoreImportQueueEvents } from "../worker/queue"; import ScoreImportFatalError from "./score-importing/score-import-error"; +import { Sleep } from "utils/misc"; +import CreateLogCtx from "lib/logger/logger"; + +const logger = CreateLogCtx(__filename); /** * Makes a score import given ScoreImportJobData. @@ -21,17 +26,45 @@ export async function MakeScoreImport( jobData: ScoreImportJobData ): Promise { if (ServerConfig.USE_EXTERNAL_SCORE_IMPORT_WORKER && process.env.IS_JOB === undefined) { - const job = await ScoreImportQueue.add(`Import ${jobData.importID}`, jobData, { - jobId: jobData.importID, - }); + let timesAttempted = 1; - const data = await job.waitUntilFinished(ScoreImportQueueEvents); + // There's no chance this thing goes on 10 times. + // if it does, this import has been trying for the past 2 days or so. + while (timesAttempted < 10) { + const job = await ScoreImportQueue.add( + `Import ${jobData.importID}${timesAttempted > 0 ? ` (TRY${timesAttempted})` : ""}`, + jobData, + { + jobId: `${jobData.importID}:TRY${timesAttempted}`, + } + ); - if (data.success) { - return data.importDocument; - } else { - throw new ScoreImportFatalError(data.statusCode, data.description); + const data = await job.waitUntilFinished(ScoreImportQueueEvents); + + if (data.success) { + return data.importDocument; + } else if (data.statusCode !== 409) { + throw new ScoreImportFatalError(data.statusCode, data.description); + } + + const backoff = ExponentialBackoff(timesAttempted - 1); + + logger.info( + `User ${jobData.userID} already had an import ongoing. (${ + jobData.importID + }) Backing off for ${(backoff / 1_000).toFixed(2)} seconds.` + ); + + // If we get here, we were 409'd and the user already has an ongoing + // import. + // In the interest of not just throwing scores away, we'll back off a bit + // and then restart the job. + await Sleep(backoff); + + timesAttempted++; } + + throw new ScoreImportFatalError(409, "Couldn't get an import at all."); } else { const InputParser = GetInputParser(jobData); @@ -44,3 +77,14 @@ export async function MakeScoreImport( ); } } + +function ExponentialBackoff(exponent: integer) { + // n | backoff + // 0 | 4 Seconds + // 1 | 16 Seconds + // 2 | 64 Seconds + // 3 | 256 Seconds + // 4 | 1024 Seconds + + return 1000 * 4 ** exponent; +} diff --git a/server/src/lib/score-import/framework/score-importing/score-import-main.ts b/server/src/lib/score-import/framework/score-importing/score-import-main.ts index cf21bc421..7d66cc583 100644 --- a/server/src/lib/score-import/framework/score-importing/score-import-main.ts +++ b/server/src/lib/score-import/framework/score-importing/score-import-main.ts @@ -1,5 +1,5 @@ import db from "external/mongo/db"; -import { KtLogger } from "lib/logger/logger"; +import { KtLogger, rootLogger } from "lib/logger/logger"; import { ScoreImportJob } from "lib/score-import/worker/types"; import { Game, @@ -12,7 +12,7 @@ import { PublicUserDocument, GetGameConfig, } from "tachi-common"; -import { GetMillisecondsSince } from "utils/misc"; +import { GetMillisecondsSince, Sleep } from "utils/misc"; import { GetUserWithID } from "utils/user"; import { ConverterFunction, ImportInputParser } from "../../import-types/common/types"; import { Converters } from "../../import-types/converters"; @@ -42,6 +42,8 @@ export default async function ScoreImportMain( providedLogger?: KtLogger, job?: ScoreImportJob ) { + rootLogger.crit(`HIT from ${userID} ${importID}.`); + const user = await GetUserWithID(userID); if (!user) { @@ -50,35 +52,38 @@ export default async function ScoreImportMain( ); } + let logger; + + if (!providedLogger) { + // If they weren't given to us - + // we create an "import logger". + // this holds a reference to the user's name, ID, and type + // of score import for any future debugging. + logger = CreateScoreLogger(user, importID, importType); + logger.debug("Received import request."); + } else { + logger = providedLogger; + } + + const hasNoOngoingImport = await CheckAndSetOngoingImportLock(user.id); + + if (hasNoOngoingImport) { + logger.info(`User ${userID} made an import while they had one ongoing.`); + // @danger + // Throwing away an import if the user already has one outgoing is *bad*, as in the case + // of degraded performance we might just start throwing scores away. + // Under normal circumstances, there is no scenario where a user would have two ongoing + // imports at the same time - even if they were using single-score imports on a 5 second + // chart, as each score import takes only around ~10-15milliseconds. + throw new ScoreImportFatalError(409, "This user already has an ongoing import."); + } + try { - const hasNoOngoingImport = await CheckAndSetOngoingImportLock(user.id); - - if (hasNoOngoingImport) { - // @danger - // Throwing away an import if the user already has one outgoing is *bad*, as in the case - // of degraded performance we might just start throwing scores away. - // Under normal circumstances, there is no scenario where a user would have two ongoing - // imports at the same time - even if they were using single-score imports on a 5 second - // chart, as each score import takes only around ~10-15milliseconds. - throw new ScoreImportFatalError(409, "This user already has an ongoing import."); - } - const timeStarted = Date.now(); - let logger; - - if (!providedLogger) { - // If they weren't given to us - - // we create an "import logger". - // this holds a reference to the user's name, ID, and type - // of score import for any future debugging. - logger = CreateScoreLogger(user, importID, importType); - logger.debug("Received import request."); - } else { - logger = providedLogger; - } SetJobProgress(job, "Parsing score data."); + await Sleep(10_000); // --- 1. Parsing --- // We get an iterable from the provided parser function, alongside some context and a converter function. // This iterable does not have to be an array - it's anything that's iterable, like a generator. @@ -240,6 +245,8 @@ export default async function ScoreImportMain( }, }); + rootLogger.crit(`END ${importID}.`); + return ImportDocument; } finally { await UnsetOngoingImportLock(user.id); From e81725df87a929f704a3e657a1235671aa3ef149 Mon Sep 17 00:00:00 2001 From: zkldi <20380519+zkldi@users.noreply.github.com> Date: Tue, 1 Feb 2022 02:02:30 +0000 Subject: [PATCH 2/3] Remove unused crit logging --- .../framework/import-locks/lock.ts | 2 -- .../lib/score-import/framework/score-import.ts | 18 ++++++++++++++---- .../score-importing/score-import-main.ts | 4 ---- 3 files changed, 14 insertions(+), 10 deletions(-) diff --git a/server/src/lib/score-import/framework/import-locks/lock.ts b/server/src/lib/score-import/framework/import-locks/lock.ts index 8d054dbdd..cb632a0ea 100644 --- a/server/src/lib/score-import/framework/import-locks/lock.ts +++ b/server/src/lib/score-import/framework/import-locks/lock.ts @@ -32,8 +32,6 @@ export async function CheckAndSetOngoingImportLock(userID: integer) { } ); - rootLogger.crit("", lockWasSet); - return !lockWasSet; } diff --git a/server/src/lib/score-import/framework/score-import.ts b/server/src/lib/score-import/framework/score-import.ts index d687ee9f3..80b2a3b5f 100644 --- a/server/src/lib/score-import/framework/score-import.ts +++ b/server/src/lib/score-import/framework/score-import.ts @@ -28,9 +28,9 @@ export async function MakeScoreImport( if (ServerConfig.USE_EXTERNAL_SCORE_IMPORT_WORKER && process.env.IS_JOB === undefined) { let timesAttempted = 1; - // There's no chance this thing goes on 10 times. - // if it does, this import has been trying for the past 2 days or so. - while (timesAttempted < 10) { + // There's no chance this thing goes on 7 times. + // if it does, this import has been trying for the past 6 hours or so. + while (timesAttempted <= 7) { const job = await ScoreImportQueue.add( `Import ${jobData.importID}${timesAttempted > 0 ? ` (TRY${timesAttempted})` : ""}`, jobData, @@ -64,7 +64,15 @@ export async function MakeScoreImport( timesAttempted++; } - throw new ScoreImportFatalError(409, "Couldn't get an import at all."); + logger.error( + `User ${jobData.userID} didn't get an import through in around 6 hours. Has their lock gotten stuck?`, + jobData + ); + + throw new ScoreImportFatalError( + 409, + "Couldn't get an import through in the past 6 hours, at all." + ); } else { const InputParser = GetInputParser(jobData); @@ -85,6 +93,8 @@ function ExponentialBackoff(exponent: integer) { // 2 | 64 Seconds // 3 | 256 Seconds // 4 | 1024 Seconds + // ... + // ends at 7, which is around 4 hours. return 1000 * 4 ** exponent; } diff --git a/server/src/lib/score-import/framework/score-importing/score-import-main.ts b/server/src/lib/score-import/framework/score-importing/score-import-main.ts index 7d66cc583..1537d1739 100644 --- a/server/src/lib/score-import/framework/score-importing/score-import-main.ts +++ b/server/src/lib/score-import/framework/score-importing/score-import-main.ts @@ -42,8 +42,6 @@ export default async function ScoreImportMain( providedLogger?: KtLogger, job?: ScoreImportJob ) { - rootLogger.crit(`HIT from ${userID} ${importID}.`); - const user = await GetUserWithID(userID); if (!user) { @@ -245,8 +243,6 @@ export default async function ScoreImportMain( }, }); - rootLogger.crit(`END ${importID}.`); - return ImportDocument; } finally { await UnsetOngoingImportLock(user.id); From 0b451d44679964ce0a06953ebe86a6bad16ef68e Mon Sep 17 00:00:00 2001 From: zkldi <20380519+zkldi@users.noreply.github.com> Date: Tue, 1 Feb 2022 02:21:05 +0000 Subject: [PATCH 3/3] Remove a sleep(10_000) --- .../score-import/framework/score-importing/score-import-main.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/server/src/lib/score-import/framework/score-importing/score-import-main.ts b/server/src/lib/score-import/framework/score-importing/score-import-main.ts index 1537d1739..0d2e24598 100644 --- a/server/src/lib/score-import/framework/score-importing/score-import-main.ts +++ b/server/src/lib/score-import/framework/score-importing/score-import-main.ts @@ -81,7 +81,6 @@ export default async function ScoreImportMain( SetJobProgress(job, "Parsing score data."); - await Sleep(10_000); // --- 1. Parsing --- // We get an iterable from the provided parser function, alongside some context and a converter function. // This iterable does not have to be an array - it's anything that's iterable, like a generator.