fix: keep transient evaluation failures alert-free, timeout-aware and idempotent across redeliveries (lfx-cdp-insights-ops#215)

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?.()

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<void> => {
     }
   }
 
-  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())
+}

services/apps/projects_evaluation_worker/src/activities/activities.ts

@@ -380,36 +380,59 @@ export async function askRequester(project: IDbProjectCatalog): Promise<void> {
 
 export type EvaluationFailureRecord = 'failed' | 'deferred' | 'moved'
 
+type FailedEvaluationRow = Pick<IDbProjectCatalog, 'id' | 'provenance' | 'evaluationAttempts'>
+
+async function isAlreadyDeferred(
+  qx: QueryExecutor,
+  project: FailedEvaluationRow,
+): Promise<boolean> {
+  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<boolean> {
-  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<EvaluationFailureRecord> {
   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'

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,

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<string> {
   try {
     const body = (await response.clone().json()) as { error?: { message?: string } }
@@ -74,6 +82,7 @@ export async function evaluateProject(input: IEvaluationInput): Promise<IEvaluat
         Authorization: `Bearer ${apiKey}`,
       },
       body: JSON.stringify(input),
+      signal: AbortSignal.timeout(EVALUATION_REQUEST_TIMEOUT_MS),
     })
   } catch (err) {
     throw transientEvaluationFailure(`API request failed: ${getErrorMessage(err)}`)
@@ -114,6 +123,9 @@ export async function evaluateProject(input: IEvaluationInput): Promise<IEvaluat
   try {
     result = await response.json()
   } catch (err) {
+    if (isAbortError(err)) {
+      throw transientEvaluationFailure(`API response stalled: ${getErrorMessage(err)}`)
+    }
     return {
       outcome: 'unsure',
       evaluationResult: 'error',

services/apps/projects_evaluation_worker/src/evaluator/retry.ts

@@ -2,6 +2,20 @@ export const TRANSIENT_EVALUATION_FAILURE = 'TransientEvaluationFailure'
 
 export const MAX_EVALUATION_RUNS = 3
 
+export const EVALUATION_ATTEMPTS = 3
+export const EVALUATION_ATTEMPT_TIMEOUT_MINUTES = 3
+export const EVALUATION_RETRY_INITIAL_SECONDS = 30
+export const EVALUATION_RETRY_BACKOFF = 2
+export const EVALUATION_REQUEST_TIMEOUT_MS = 150_000
+
+export function evaluationWorstCaseMinutesPerProject(): number {
+  let backoffSeconds = 0
+  for (let attempt = 1; attempt < EVALUATION_ATTEMPTS; attempt++) {
+    backoffSeconds += EVALUATION_RETRY_INITIAL_SECONDS * EVALUATION_RETRY_BACKOFF ** (attempt - 1)
+  }
+  return Math.ceil(EVALUATION_ATTEMPTS * EVALUATION_ATTEMPT_TIMEOUT_MINUTES + backoffSeconds / 60)
+}
+
 export function isTransientHttpStatus(status: number): boolean {
   return status === 429 || status >= 500
 }

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 = {

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<typeof activities>({
 
 // Each AI evaluation call takes ~30-40s; give generous headroom per project.
 const evaluateActivities = proxyActivities<typeof activities>({
-  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<typeof activities>({
@@ -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 {

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)
   })