From 7b67b077dd9523ac8e32c39522d96d13aceaa07d Mon Sep 17 00:00:00 2001 From: kelvin <43873157+kelvinkipruto@users.noreply.github.com> Date: Mon, 10 Aug 2026 18:46:53 +0300 Subject: [PATCH 1/2] fix: stop treating CheckMedia 429s as permanent upload failures Rate-limited extractions were being written to uploadError and never retried, so a single burst of requests could permanently strand a large chunk of the upload backlog. Uploads are now paced, 429/5xx/network errors are left pending for the next scheduled run instead of being poisoned, and a run stops early once CheckMedia signals it's rate limited rather than cascading through the rest of the batch. Also adds a one-off script to clear uploadError on extractions that were already poisoned by a 429 before this fix, and a manual workflow_dispatch job to run maintenance scripts like it against a chosen environment via the existing migrator image. Co-Authored-By: Claude Sonnet 5 --- .github/workflows/run-maintenance-script.yml | 71 ++++++++++++ scripts/clear-checkmedia-rate-limit-errors.ts | 82 ++++++++++++++ src/lib/meedan.ts | 103 ++++++++++++++---- src/tasks/uploadToMeedan.ts | 96 +++++++++++++++- 4 files changed, 326 insertions(+), 26 deletions(-) create mode 100644 .github/workflows/run-maintenance-script.yml create mode 100644 scripts/clear-checkmedia-rate-limit-errors.ts diff --git a/.github/workflows/run-maintenance-script.yml b/.github/workflows/run-maintenance-script.yml new file mode 100644 index 00000000..6ee0237f --- /dev/null +++ b/.github/workflows/run-maintenance-script.yml @@ -0,0 +1,71 @@ +name: Run maintenance script + +# Manually triggered one-off script runner. Runs a script from scripts/ inside +# the migrator image (full source + Payload CLI + node_modules) against a +# chosen environment's database — the same mechanism promote.yml uses to run +# `payload migrate` during a release, reused here for one-off data fixes that +# don't belong in a migration (see scripts/*.ts for examples). +# +# Defaults to a dry run: the script only writes data when "Apply changes" is +# checked. Targeting "production" is gated by the `production` GitHub +# environment's required reviewers, same as a real release. +on: + workflow_dispatch: + inputs: + environment: + description: "Target environment (selects which database the script runs against)" + type: choice + options: + - development + - staging + - production + required: true + script_path: + description: "Script path relative to repo root" + type: string + required: true + default: "scripts/clear-checkmedia-rate-limit-errors.ts" + apply: + description: "Apply changes (unchecked = dry run)" + type: boolean + default: false + +concurrency: + group: "run-maintenance-script @ ${{ inputs.environment }}" + cancel-in-progress: false + +# Least-privilege default for GITHUB_TOKEN; the run authenticates to +# DockerHub via dedicated secrets, not the workflow token. +permissions: + contents: read + +jobs: + run: + runs-on: ubuntu-latest + environment: + name: ${{ inputs.environment }} + steps: + - name: Login to DockerHub + uses: docker/login-action@v4 + with: + username: ${{ secrets.DOCKER_HUB_USERNAME }} + password: ${{ secrets.DOCKER_HUB_ACCESS_TOKEN }} + + # Runs the migrator image built for the commit this workflow was + # dispatched against — trigger this from the same ref (branch or tag) + # that was actually deployed, so the script sees the code it was + # written for. Staging uses distinctly-named secrets (no GitHub + # Environment override configured for it); development and production + # share secret names but resolve to different values via their + # respective GitHub Environments. + - name: Run script against ${{ inputs.environment }} + env: + DATABASE_URI: ${{ inputs.environment == 'staging' && secrets.STAGING_DATABASE_URI || secrets.DATABASE_URI }} + PAYLOAD_SECRET: ${{ inputs.environment == 'staging' && secrets.STAGING_PAYLOAD_SECRET || secrets.PAYLOAD_SECRET }} + run: | + docker run --rm \ + -e DATABASE_URI \ + -e PAYLOAD_SECRET \ + -e TIKA_ENABLED=0 \ + "codeforafrica/promisetracker-v2-migrator:${{ github.sha }}" \ + pnpm payload run "${{ inputs.script_path }}" ${{ inputs.apply && '-- --apply' || '' }} diff --git a/scripts/clear-checkmedia-rate-limit-errors.ts b/scripts/clear-checkmedia-rate-limit-errors.ts new file mode 100644 index 00000000..6c6425fa --- /dev/null +++ b/scripts/clear-checkmedia-rate-limit-errors.ts @@ -0,0 +1,82 @@ +/** + * One-off remediation for the pre-fix uploadToMeedan behavior that treated + * CheckMedia HTTP 429 (rate limit) responses as permanent failures. Clears + * `uploadError` on extractions whose stored error is a 429 so they become + * eligible for upload again on the next uploadToMeedan run. + * + * Non-429 errors (bad source URLs, GraphQL validation errors, missing + * entity/tenant, etc.) are left untouched — those are deterministic + * failures that would just fail again on retry. + * + * Dry run (default): + * pnpm payload run scripts/clear-checkmedia-rate-limit-errors.ts + * Apply: + * pnpm payload run scripts/clear-checkmedia-rate-limit-errors.ts -- --apply + */ +import { getPayload } from "payload"; +import config from "../src/payload.config"; + +const apply = process.argv.includes("--apply"); + +const isRateLimitError = (message: string | null | undefined): boolean => + Boolean(message) && /HTTP 429|TooManyRequestsError/i.test(message as string); + +const payload = await getPayload({ config }); + +const { docs: extractionDocs } = await payload.find({ + collection: "ai-extractions", + depth: 0, + limit: 0, +}); + +console.log( + `Found ${extractionDocs.length} ai-extractions docs (apply=${apply})`, +); + +let clearedExtractions = 0; +let touchedDocs = 0; + +for (const doc of extractionDocs) { + const extractions = doc.extractions ?? []; + const matches = extractions.filter((extraction) => + isRateLimitError(extraction.uploadError), + ); + + if (matches.length === 0) { + continue; + } + + touchedDocs += 1; + for (const match of matches) { + console.log( + JSON.stringify({ + action: apply ? "clear" : "would-clear", + extractionDocId: doc.id, + extractionUniqueId: match.uniqueId, + uploadError: match.uploadError, + }), + ); + } + + if (apply) { + const updatedExtractions = extractions.map((extraction) => + isRateLimitError(extraction.uploadError) + ? { ...extraction, uploadError: null } + : extraction, + ); + await payload.update({ + collection: "ai-extractions", + id: doc.id, + data: { extractions: updatedExtractions }, + }); + } + + clearedExtractions += matches.length; +} + +console.log( + `Done. ${apply ? "Cleared" : "Would clear"} ${clearedExtractions} extraction(s) across ${touchedDocs} document(s).`, +); + +await payload.destroy(); +process.exit(0); diff --git a/src/lib/meedan.ts b/src/lib/meedan.ts index d192b0ba..b028bf10 100644 --- a/src/lib/meedan.ts +++ b/src/lib/meedan.ts @@ -43,6 +43,45 @@ const PUBLISHED_REPORTS_QUERY = ` const toErrorMessage = (error: unknown): string => error instanceof Error ? error.message : String(error); +// Thrown for any non-OK HTTP response from CheckMedia. Callers use `status` +// to distinguish transient failures (429, 5xx) from permanent ones so a +// rate limit doesn't get treated the same as a validation error. +export class CheckMediaHttpError extends Error { + readonly status: number; + readonly retryAfterMs: number | null; + + constructor(message: string, status: number, retryAfterMs: number | null) { + super(message); + this.name = "CheckMediaHttpError"; + this.status = status; + this.retryAfterMs = retryAfterMs; + } +} + +// Parses a `Retry-After` header per RFC 9110: either a number of seconds or +// an HTTP-date. Returns null when absent or unparseable. +const parseRetryAfterMs = (response: Response): number | null => { + const header = response.headers.get("retry-after"); + if (!header) { + return null; + } + + const seconds = Number(header); + if (Number.isFinite(seconds)) { + return Math.max(0, seconds * 1000); + } + + const dateMs = Date.parse(header); + if (!Number.isNaN(dateMs)) { + return Math.max(0, dateMs - Date.now()); + } + + return null; +}; + +const sleep = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + const previewResponseBody = (rawBody: string): string => { const normalized = rawBody.replace(/\s+/g, " ").trim(); @@ -90,8 +129,10 @@ const throwHttpErrorWithBody = async ({ } const traceHeaders = getTraceHeaders(response); - throw new Error( + throw new CheckMediaHttpError( `[CheckMedia:${operation}] HTTP ${response.status} ${response.statusText}; traceHeaders=${traceHeaders}; responseBody=${bodyPreview}`, + response.status, + parseRetryAfterMs(response), ); }; @@ -529,6 +570,14 @@ export const fetchPublishedReports = async ({ return mapPublishedReports(json); }; +// Bounded inline retry for HTTP 429 only. CheckMedia's rate-limit window is +// short (seconds), so it's worth waiting once or twice in-process before +// giving up; anything longer than MAX_RATE_LIMIT_WAIT_MS is left to the next +// scheduled task run instead of blocking this one. +const MAX_RATE_LIMIT_RETRIES = 2; +const DEFAULT_RATE_LIMIT_BACKOFF_MS = 2000; +const MAX_RATE_LIMIT_WAIT_MS = 10_000; + export const postRequest = async ({ apiKey, teamId, @@ -548,32 +597,44 @@ export const postRequest = async ({ }; try { - const response = await fetch(BASE_URL, { - method: "POST", - headers: { - "Content-Type": "application/json", - "X-Check-Token": apiKey, - "X-Check-Team": teamId, - }, - body: JSON.stringify(requestBody), - }); + for (let attempt = 0; ; attempt += 1) { + const response = await fetch(BASE_URL, { + method: "POST", + headers: { + "Content-Type": "application/json", + "X-Check-Token": apiKey, + "X-Check-Team": teamId, + }, + body: JSON.stringify(requestBody), + }); + + if (response.ok) { + const result: CreateProjectMediaResponse = await response.json(); + + if (result.errors && result.errors.length > 0) { + throw new Error( + `[CheckMedia:postRequest.createProjectMedia] GraphQL errors: ${formatGraphQLErrorDetails(result.errors)}`, + ); + } + + return result; + } + + if (response.status === 429 && attempt < MAX_RATE_LIMIT_RETRIES) { + const retryAfterMs = parseRetryAfterMs(response); + const waitMs = Math.min( + retryAfterMs ?? DEFAULT_RATE_LIMIT_BACKOFF_MS * (attempt + 1), + MAX_RATE_LIMIT_WAIT_MS, + ); + await sleep(waitMs); + continue; + } - if (!response.ok) { await throwHttpErrorWithBody({ response, operation: "postRequest.createProjectMedia", }); } - - const result: CreateProjectMediaResponse = await response.json(); - - if (result.errors && result.errors.length > 0) { - throw new Error( - `[CheckMedia:postRequest.createProjectMedia] GraphQL errors: ${formatGraphQLErrorDetails(result.errors)}`, - ); - } - - return result; } catch (error) { console.error("Error making request to CheckMedia:", error); throw error; diff --git a/src/tasks/uploadToMeedan.ts b/src/tasks/uploadToMeedan.ts index fcfcc26f..d7249255 100644 --- a/src/tasks/uploadToMeedan.ts +++ b/src/tasks/uploadToMeedan.ts @@ -1,5 +1,5 @@ import { TaskConfig } from "payload"; -import { createFactCheckClaim } from "@/lib/meedan"; +import { CheckMediaHttpError, createFactCheckClaim } from "@/lib/meedan"; import { markDocumentAsProcessed, updateDocumentStatus } from "@/lib/airtable"; import { AiExtraction as AiExtractionDoc, @@ -149,6 +149,28 @@ const resolveCheckMediaSourceUrl = ({ ); }; +const sleep = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +// Spacing between successive CheckMedia upload calls so a large backlog +// doesn't burst past CheckMedia's rate limit in the first place. +const UPLOAD_REQUEST_DELAY_MS = Number( + process.env.MEEDAN_UPLOAD_REQUEST_DELAY_MS ?? 350, +); + +// 429 (rate limited) and 5xx (CheckMedia-side errors) are treated as +// transient: the extraction is left pending instead of being marked +// `uploadError`, so it's retried automatically on the next run. Network +// failures (fetch rejecting before a response is received) are transient +// too. Everything else (4xx other than 429, GraphQL errors, bad source +// URLs) is a genuine, deterministic failure that won't succeed on retry. +const isRetryableUploadError = (error: unknown): boolean => { + if (error instanceof CheckMediaHttpError) { + return error.status === 429 || error.status >= 500; + } + return error instanceof TypeError; +}; + const hasPendingExtractions = (doc: AiExtractionDoc): boolean => (doc.extractions ?? []).some( (extraction) => !extraction?.checkMediaId && !extraction?.uploadError, @@ -292,10 +314,13 @@ export const UploadToMeedan: TaskConfig<"uploadToMeedan"> = { let hasNextPage = true; let uploadedExtractions = 0; let failedExtractions = 0; + let deferredExtractions = 0; let processedExtractionDocs = 0; let failedExtractionDocs = 0; + let hasAttemptedUpload = false; + let stopUploadingDueToRateLimit = false; - while (hasNextPage) { + while (hasNextPage && !stopUploadingDueToRateLimit) { const { docs: allExtractions, hasNextPage: nextPage } = await payload.find({ collection: "ai-extractions", @@ -305,6 +330,10 @@ export const UploadToMeedan: TaskConfig<"uploadToMeedan"> = { }); for (const doc of allExtractions) { + if (stopUploadingDueToRateLimit) { + break; + } + const document = doc.document as Document; const documentId = document?.id ? String(document.id) : undefined; @@ -415,9 +444,19 @@ export const UploadToMeedan: TaskConfig<"uploadToMeedan"> = { } let docFailedExtractions = 0; let docUploadedExtractions = 0; + let docDeferredExtractions = 0; for (const extraction of extractionsToUpload) { + if (stopUploadingDueToRateLimit) { + break; + } + try { + if (hasAttemptedUpload) { + await sleep(UPLOAD_REQUEST_DELAY_MS); + } + hasAttemptedUpload = true; + logger.info({ message: "uploadToMeedan:: Uploading extraction to CheckMedia", extractionDocId: doc.id, @@ -543,13 +582,43 @@ export const UploadToMeedan: TaskConfig<"uploadToMeedan"> = { checkMediaURL, }); } catch (extractionError) { - failedExtractions += 1; - docFailedExtractions += 1; const uploadErrorMessage = extractionError instanceof Error ? extractionError.message : String(extractionError); + if (isRetryableUploadError(extractionError)) { + deferredExtractions += 1; + docDeferredExtractions += 1; + + const isRateLimited = + extractionError instanceof CheckMediaHttpError && + extractionError.status === 429; + + logger.warn({ + message: isRateLimited + ? "uploadToMeedan:: CheckMedia rate limit hit — leaving extraction pending and deferring remaining uploads to the next run" + : "uploadToMeedan:: Upload failed with a transient error — left pending for retry on next run", + extractionDocId: doc.id, + extractionDocTitle: doc.title, + extractionUniqueId: extraction.uniqueId, + documentId, + documentTitle: document?.title, + documentAirtableID: document?.airtableID, + error: uploadErrorMessage, + }); + + if (isRateLimited) { + stopUploadingDueToRateLimit = true; + break; + } + + continue; + } + + failedExtractions += 1; + docFailedExtractions += 1; + // Re-fetch before writing so we don't clobber checkMediaId or // uploadError values written by earlier iterations in this loop. const freshDoc = await payload.findByID({ @@ -603,6 +672,19 @@ export const UploadToMeedan: TaskConfig<"uploadToMeedan"> = { }); } + if (docDeferredExtractions > 0) { + logger.warn({ + message: `uploadToMeedan:: ${docDeferredExtractions} out of ${doc.extractions?.length ?? 0} extraction(s) deferred due to a transient error — will retry on next run`, + extractionDocId: doc.id, + extractionDocTitle: doc.title, + documentId, + documentTitle: document?.title, + documentAirtableID: document?.airtableID, + deferredExtractions: docDeferredExtractions, + totalExtractionsToUpload: extractionsToUpload.length, + }); + } + const totalExtractions = doc.extractions?.length ?? 0; const completionStatus = docFailedExtractions > 0 @@ -670,11 +752,15 @@ export const UploadToMeedan: TaskConfig<"uploadToMeedan"> = { } logger.info({ - message: "uploadToMeedan:: Upload task completed", + message: stopUploadingDueToRateLimit + ? "uploadToMeedan:: Upload task stopped early after hitting CheckMedia's rate limit — remaining extractions will be retried on the next run" + : "uploadToMeedan:: Upload task completed", processedExtractionDocs, failedExtractionDocs, uploadedExtractions, failedExtractions, + deferredExtractions, + stoppedForRateLimit: stopUploadingDueToRateLimit, }); return { From afca353e6f1c0fd22beb9046457bc5fcdd6351be Mon Sep 17 00:00:00 2001 From: kelvin <43873157+kelvinkipruto@users.noreply.github.com> Date: Mon, 10 Aug 2026 19:03:53 +0300 Subject: [PATCH 2/2] feat: support ?queue= and ?all= params on /api/run-jobs The manual job-runner endpoint only ever ran Payload's default queue, so there was no way to nudge queues like exportSync (used by AI_EXTRACTION_EXPORT_ROWS_SYNC_CRON_SCHEDULE) without waiting for their own autoRun tick. Co-Authored-By: Claude Sonnet 5 --- src/app/api/run-jobs/route.ts | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/src/app/api/run-jobs/route.ts b/src/app/api/run-jobs/route.ts index 3c39538c..6b81ea72 100644 --- a/src/app/api/run-jobs/route.ts +++ b/src/app/api/run-jobs/route.ts @@ -9,6 +9,11 @@ * - You are debugging a specific queue and want an immediate execution cycle. * - The everyMinute cron has not fired yet after a fresh deploy. * + * Query params (both optional; `all` takes precedence over `queue`): + * - `queue`: run jobs from this queue instead of Payload's default `default` + * queue (e.g. `?queue=exportSync`). + * - `all=true`: run jobs from every queue, ignoring `queue`. + * * Security: requires an active Payload CMS session (i.e. the caller must be * logged in to the admin panel). Unauthenticated requests are rejected with * HTTP 401. @@ -30,12 +35,23 @@ export const GET = async (request: NextRequest) => { return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); } - payload.jobs.run().catch((error: unknown) => { - payload.logger.error({ - msg: "run-jobs:: Unhandled error in jobs.run()", - error, + const { searchParams } = new URL(request.url); + const allQueues = searchParams.get("all") === "true"; + const queue = searchParams.get("queue") || undefined; + + payload.jobs + .run(allQueues ? { allQueues: true } : { queue }) + .catch((error: unknown) => { + payload.logger.error({ + msg: "run-jobs:: Unhandled error in jobs.run()", + allQueues, + queue, + error, + }); }); - }); - return NextResponse.json({ ok: true }, { status: 202 }); + return NextResponse.json( + { ok: true, allQueues, queue: allQueues ? undefined : (queue ?? "default") }, + { status: 202 }, + ); };