diff --git a/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.test.ts b/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.test.ts index 24e29f883a..2e8f498d48 100644 --- a/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.test.ts +++ b/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.test.ts @@ -1,6 +1,7 @@ import type { Request, Response } from 'express' import { beforeEach, describe, expect, it, vi } from 'vitest' +import { ServiceUnavailableError } from '@crowd/common' import { PRECHECK_SKIP_REASONS } from '@crowd/data-access-layer/src/project-catalog/precheck' import { IProjectEvaluationResponse } from '@crowd/data-access-layer/src/project-catalog/types' @@ -74,6 +75,27 @@ describe('projectEvaluationExec', () => { expect(json).toHaveBeenCalledWith(evaluationResponse) }) + it('answers a transient evaluator failure with 503 without raising a public API alert', async () => { + vi.mocked(evaluateProject).mockRejectedValue(new ServiceUnavailableError('Bedrock throttled')) + + const { req, res, status, json } = mockReqRes({ + id: 'catalog-1', + repoUrl: 'https://github.com/foo/bar', + repoName: 'bar', + projectSlug: 'foo', + lfCriticalityScore: null, + source: null, + }) + + await expect(projectEvaluationExec(req, res)).resolves.toBeUndefined() + + expect(status).toHaveBeenCalledWith(503) + expect(json).toHaveBeenCalledWith({ + error: { code: 'SERVICE_UNAVAILABLE', message: 'Bedrock throttled' }, + }) + expect(req.log.warn).toHaveBeenCalled() + }) + it('keys the daily LLM reservation on the actor id', async () => { vi.mocked(evaluateProject).mockImplementation(async (_input, _qx, _creds, _log, reserve) => { reserve?.() diff --git a/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.ts b/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.ts index e070bf013f..56cc1e6168 100644 --- a/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.ts +++ b/backend/src/api/public/v1/projectEvaluation/projectEvaluationExec.ts @@ -3,11 +3,16 @@ import type { Request, Response } from 'express' import { optionsQx } from '@/database/sequelizeQueryExecutor' import { ok } from '@/utils/api' import { validateOrThrow } from '@/utils/validation' +import { ServiceUnavailableError } from '@crowd/common' import { createDailyLlmCap } from './dailyLlmCap' import { evaluateProject } from './evaluateProject' import { precheckSkipResult, runPrecheck } from './runPrecheck' -import { IProjectEvaluationRequest, projectEvaluationRequestSchema } from './types' +import { + IProjectEvaluationRequest, + IProjectEvaluationResponse, + projectEvaluationRequestSchema, +} from './types' const reserveDailyLlmCall = createDailyLlmCap() @@ -32,16 +37,40 @@ export default async (req: Request, res: Response): Promise => { } } - const response = await evaluateProject( - input, - qx, - { - accessKeyId: process.env.CROWD_AWS_BEDROCK_ACCESS_KEY_ID, - secretAccessKey: process.env.CROWD_AWS_BEDROCK_SECRET_ACCESS_KEY, - }, - req.log, - () => reserveDailyLlmCall(req.actor.apiKeyId ?? req.actor.id, req.actor.id), - ) + let response: IProjectEvaluationResponse + try { + response = await evaluateProject( + input, + qx, + { + accessKeyId: process.env.CROWD_AWS_BEDROCK_ACCESS_KEY_ID, + secretAccessKey: process.env.CROWD_AWS_BEDROCK_SECRET_ACCESS_KEY, + }, + req.log, + () => reserveDailyLlmCall(req.actor.apiKeyId ?? req.actor.id, req.actor.id), + ) + } catch (err) { + if (err instanceof ServiceUnavailableError) { + respondTransientFailure(req, res, input, err) + return + } + throw err + } ok(res, response) } + +// Answered here instead of the error handler: the worker retries these, so the +// public-API 5xx alert would fire for failures that end well. +function respondTransientFailure( + req: Request, + res: Response, + input: IProjectEvaluationRequest, + err: ServiceUnavailableError, +): void { + req.log.warn( + { id: input.id, repoUrl: input.repoUrl, reason: err.message }, + 'Project evaluation hit a transient error, the worker will retry.', + ) + res.status(err.status).json(err.toJSON()) +} diff --git a/services/apps/projects_evaluation_worker/src/activities/activities.ts b/services/apps/projects_evaluation_worker/src/activities/activities.ts index 07b64f15ba..de071ffc17 100644 --- a/services/apps/projects_evaluation_worker/src/activities/activities.ts +++ b/services/apps/projects_evaluation_worker/src/activities/activities.ts @@ -380,36 +380,59 @@ export async function askRequester(project: IDbProjectCatalog): Promise { export type EvaluationFailureRecord = 'failed' | 'deferred' | 'moved' +type FailedEvaluationRow = Pick + +async function isAlreadyDeferred( + qx: QueryExecutor, + project: FailedEvaluationRow, +): Promise { + const current = await findProjectCatalogById(qx, project.id) + return ( + current?.action === 'evaluate' && + current.evaluatedAt === null && + current.evaluationAttempts === project.evaluationAttempts + 1 + ) +} + async function deferEvaluation( qx: QueryExecutor, - projectId: string, + project: FailedEvaluationRow, reason: string, ): Promise { - const current = await findProjectCatalogById(qx, projectId) - if (!current || !canDeferEvaluation(current)) { + if (!canDeferEvaluation(project)) { return false } - const deferred = await deferProjectCatalogEvaluation(qx, projectId, reason, MAX_EVALUATION_RUNS) + const deferred = await deferProjectCatalogEvaluation(qx, project.id, { + reason, + expectedAttempts: project.evaluationAttempts, + maxRuns: MAX_EVALUATION_RUNS, + }) if (!deferred) { - return false + return isAlreadyDeferred(qx, project) } log.warn( - { id: projectId, reason, attempts: deferred.evaluationAttempts }, + { id: project.id, reason, attempts: deferred.evaluationAttempts }, 'Evaluation hit a transient error, the row waits for the next run.', ) return true } export async function markEvaluationFailed( - projectId: string, + project: FailedEvaluationRow, reason: string, transient = false, ): Promise { const qx = pgpQx(svc.postgres.writer.connection()) - if (transient && (await deferEvaluation(qx, projectId, reason))) { + const projectId = project.id + if (transient && (await deferEvaluation(qx, project, reason))) { return 'deferred' } - const marked = await markProjectCatalogEvaluationFailed(qx, projectId, reason) + const marked = await markProjectCatalogEvaluationFailed( + qx, + projectId, + reason, + transient ? project.evaluationAttempts : undefined, + ) if (marked > 0) { log.error({ id: projectId, reason }, 'Evaluation failed, marked as error.') return 'failed' diff --git a/services/apps/projects_evaluation_worker/src/evaluator/evaluator.test.ts b/services/apps/projects_evaluation_worker/src/evaluator/evaluator.test.ts index 5d034f545f..decde3c058 100644 --- a/services/apps/projects_evaluation_worker/src/evaluator/evaluator.test.ts +++ b/services/apps/projects_evaluation_worker/src/evaluator/evaluator.test.ts @@ -62,6 +62,7 @@ describe('evaluateProject', () => { Authorization: 'Bearer test-key', }, body: JSON.stringify(input), + signal: expect.any(AbortSignal), }) expect(result).toEqual(apiResult) }) @@ -165,6 +166,25 @@ describe('evaluateProject', () => { expect(result.evaluationReason).toContain('400') }) + it('throws a transient failure when the response body stalls past the deadline', async () => { + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + statusText: 'OK', + json: async () => { + throw Object.assign(new Error('The operation was aborted due to timeout'), { + name: 'TimeoutError', + }) + }, + }) + + await expect(evaluateProject(input)).rejects.toMatchObject({ + type: TRANSIENT_EVALUATION_FAILURE, + nonRetryable: false, + message: 'API response stalled: The operation was aborted due to timeout', + }) + }) + it('returns an unsure/error result when the response body is not valid JSON', async () => { fetchMock.mockResolvedValue({ ok: true, diff --git a/services/apps/projects_evaluation_worker/src/evaluator/evaluator.ts b/services/apps/projects_evaluation_worker/src/evaluator/evaluator.ts index 07f1c2297f..94bc821c3f 100644 --- a/services/apps/projects_evaluation_worker/src/evaluator/evaluator.ts +++ b/services/apps/projects_evaluation_worker/src/evaluator/evaluator.ts @@ -2,7 +2,11 @@ import { ApplicationFailure } from '@temporalio/activity' import { getErrorMessage } from '@crowd/common' -import { TRANSIENT_EVALUATION_FAILURE, isTransientHttpStatus } from './retry' +import { + EVALUATION_REQUEST_TIMEOUT_MS, + TRANSIENT_EVALUATION_FAILURE, + isTransientHttpStatus, +} from './retry' import { IEvaluationInput, IEvaluationMetrics, IEvaluationResult } from './types' export function transientEvaluationFailure( @@ -42,6 +46,10 @@ function isValidResult(result: unknown): result is IEvaluationResult { ) } +function isAbortError(err: unknown): boolean { + return err instanceof Error && (err.name === 'AbortError' || err.name === 'TimeoutError') +} + async function responseErrorMessage(response: Response): Promise { try { const body = (await response.clone().json()) as { error?: { message?: string } } @@ -74,6 +82,7 @@ export async function evaluateProject(input: IEvaluationInput): Promise= 500 } diff --git a/services/apps/projects_evaluation_worker/src/schedules/scheduleProjectsEvaluation.ts b/services/apps/projects_evaluation_worker/src/schedules/scheduleProjectsEvaluation.ts index e790585142..4de7b51f19 100644 --- a/services/apps/projects_evaluation_worker/src/schedules/scheduleProjectsEvaluation.ts +++ b/services/apps/projects_evaluation_worker/src/schedules/scheduleProjectsEvaluation.ts @@ -2,6 +2,7 @@ import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/clien import { DISCOVERY_NEW_PROJECTS_LIMIT } from '@crowd/common' +import { evaluationWorstCaseMinutesPerProject } from '../evaluator/retry' import { svc } from '../main' import { IEvaluateProjectsInput, evaluateProjects } from '../workflows' @@ -12,9 +13,7 @@ const SOURCE_PRIORITY = ['manual', 'insights-discussions', 'lf-criticality-score // DISCOVERY_NEW_PROJECTS_LIMIT per source per run, so evaluation must drain that much. const EVALUATION_LIMIT = DISCOVERY_NEW_PROJECTS_LIMIT * SOURCE_PRIORITY.length -// Worst case per project: evaluateActivities allows 2 attempts × 3 min startToCloseTimeout. -const WORST_CASE_MINUTES_PER_PROJECT = 6 -const WORKFLOW_EXECUTION_TIMEOUT = `${EVALUATION_LIMIT * WORST_CASE_MINUTES_PER_PROJECT} minutes` +const WORKFLOW_EXECUTION_TIMEOUT = `${EVALUATION_LIMIT * evaluationWorstCaseMinutesPerProject()} minutes` function scheduleAction() { const args: IEvaluateProjectsInput = { diff --git a/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts b/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts index f5ec36506e..1bddd3c261 100644 --- a/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts +++ b/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts @@ -1,6 +1,7 @@ import { ApplicationFailure, CancellationScope, + TimeoutFailure, isCancellation, log, patched, @@ -11,7 +12,13 @@ import { import type { IPipelineRunEvaluatorModelUsage } from '@crowd/data-access-layer/src/project-catalog-pipeline-runs/types' import type * as activities from '../activities' -import { TRANSIENT_EVALUATION_FAILURE } from '../evaluator/retry' +import { + EVALUATION_ATTEMPTS, + EVALUATION_ATTEMPT_TIMEOUT_MINUTES, + EVALUATION_RETRY_BACKOFF, + EVALUATION_RETRY_INITIAL_SECONDS, + TRANSIENT_EVALUATION_FAILURE, +} from '../evaluator/retry' import type { IClassifyAndRouteResult, IEvaluateProjectsInput, @@ -41,8 +48,12 @@ const classifyActivities = proxyActivities({ // Each AI evaluation call takes ~30-40s; give generous headroom per project. const evaluateActivities = proxyActivities({ - startToCloseTimeout: '3 minutes', - retry: { maximumAttempts: 3, initialInterval: '30 seconds', backoffCoefficient: 2 }, + startToCloseTimeout: `${EVALUATION_ATTEMPT_TIMEOUT_MINUTES} minutes`, + retry: { + maximumAttempts: EVALUATION_ATTEMPTS, + initialInterval: `${EVALUATION_RETRY_INITIAL_SECONDS} seconds`, + backoffCoefficient: EVALUATION_RETRY_BACKOFF, + }, }) const pipelineRunActivities = proxyActivities({ @@ -108,6 +119,9 @@ function rootErrorMessage(err: unknown): string { function isTransientEvaluationFailure(err: unknown): boolean { let current = err while (current instanceof Error) { + if (current instanceof TimeoutFailure) { + return true + } if (current instanceof ApplicationFailure && current.type === TRANSIENT_EVALUATION_FAILURE) { return true } @@ -285,11 +299,7 @@ export async function evaluateProjects(input: IEvaluateProjectsInput = {}): Prom `Evaluation failed for project id=${project.id} repoUrl=${project.repoUrl}: ${reason}`, ) try { - const recorded = await markActivities.markEvaluationFailed( - project.id, - reason, - transient, - ) + const recorded = await markActivities.markEvaluationFailed(project, reason, transient) if (recorded === 'deferred') { deferred++ } else { diff --git a/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts b/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts index e63b4e5d97..0ff1f6bc83 100644 --- a/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts +++ b/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts @@ -760,19 +760,22 @@ describe('evaluation failures and retries', () => { test('defers a transient failure to the next run until the last allowed run', async ({ qx }) => { const inserted = await insertProjectCatalog(qx, catalogRow({ action: 'evaluate' })) - const first = await deferProjectCatalogEvaluation(qx, inserted.id, 'API returned HTTP 503', 3) - expect(first).toMatchObject({ + const defer = (expectedAttempts: number, reason: string) => + deferProjectCatalogEvaluation(qx, inserted.id, { reason, expectedAttempts, maxRuns: 3 }) + + expect(await defer(0, 'API returned HTTP 503')).toMatchObject({ action: 'evaluate', evaluationAttempts: 1, evaluationReason: 'API returned HTTP 503', evaluatedAt: null, }) - expect(await deferProjectCatalogEvaluation(qx, inserted.id, 'again', 3)).toMatchObject({ - evaluationAttempts: 2, - }) - expect(await deferProjectCatalogEvaluation(qx, inserted.id, 'and again', 3)).toBeNull() + expect(await defer(0, 'redelivered')).toBeNull() + expect(await defer(1, 'again')).toMatchObject({ evaluationAttempts: 2 }) + expect(await defer(2, 'and again')).toBeNull() + + expect(await markProjectCatalogEvaluationFailed(qx, inserted.id, 'stale delivery', 1)).toBe(0) + expect(await markProjectCatalogEvaluationFailed(qx, inserted.id, 'and again', 2)).toBe(1) - await markProjectCatalogEvaluationFailed(qx, inserted.id, 'and again') const requeued = await requeueProjectCatalogEvaluation(qx, inserted.id) expect(requeued?.evaluationAttempts).toBe(0) }) diff --git a/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts b/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts index daddfee036..7cad470f12 100644 --- a/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts +++ b/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts @@ -1013,6 +1013,7 @@ export async function markProjectCatalogEvaluationFailed( qx: QueryExecutor, id: string, reason: string, + expectedAttempts?: number, ): Promise { return qx.result( ` @@ -1020,8 +1021,9 @@ export async function markProjectCatalogEvaluationFailed( SET "action" = 'error', "evaluationResult" = 'error', "evaluationReason" = $(reason), "evaluatedAt" = NOW(), "updatedAt" = NOW() WHERE id = $(id) AND "action" = 'evaluate' + AND ($(expectedAttempts) IS NULL OR "evaluationAttempts" = $(expectedAttempts)) `, - { id, reason }, + { id, reason, expectedAttempts: expectedAttempts ?? null }, ) } @@ -1042,22 +1044,28 @@ export async function requeueProjectCatalogEvaluation( ) } +export interface IDbProjectCatalogDeferral { + reason: string + expectedAttempts: number + maxRuns: number +} + +// The expected counter makes a redelivered deferral a no-op instead of a second attempt. export async function deferProjectCatalogEvaluation( qx: QueryExecutor, id: string, - reason: string, - maxRuns: number, + { reason, expectedAttempts, maxRuns }: IDbProjectCatalogDeferral, ): Promise { return qx.selectOneOrNone( ` UPDATE "projectCatalog" - SET "evaluationAttempts" = "evaluationAttempts" + 1, "evaluationReason" = $(reason), + SET "evaluationAttempts" = $(expectedAttempts) + 1, "evaluationReason" = $(reason), "updatedAt" = NOW() WHERE id = $(id) AND "action" = 'evaluate' AND "evaluatedAt" IS NULL - AND "evaluationAttempts" + 1 < $(maxRuns) + AND "evaluationAttempts" = $(expectedAttempts) AND $(expectedAttempts) + 1 < $(maxRuns) RETURNING ${prepareSelectColumns(PROJECT_CATALOG_COLUMNS)} `, - { id, reason, maxRuns }, + { id, reason, expectedAttempts, maxRuns }, ) }