From c0a092142c81f6cdac55d01a3454b79823cebcb8 Mon Sep 17 00:00:00 2001 From: Wasim Amiri <7220175+wasimxyz@users.noreply.github.com> Date: Thu, 20 Aug 2026 16:25:00 -0700 Subject: [PATCH 1/2] Skip processed artifacts when reprocessing a run. Processed posters and videos often match watcher filename patterns, so run reprocess queued them and left them stuck in processing. Co-authored-by: Cursor --- developer-docs/lambda.md | 4 +- .../api/v1/files/[fileId]/reprocess/route.ts | 6 +- .../runs/[runId]/reprocess/route.ts | 4 +- .../runs/file-selection-provider.tsx | 10 +- web/components/runs/run-files-table.tsx | 22 +--- web/lib/api/file-reprocessing.ts | 14 +++ web/lib/api/openapi/paths/files.ts | 2 +- web/lib/api/openapi/paths/runs.ts | 2 +- web/lib/mcp/tools/files.defs.ts | 2 +- web/lib/mcp/tools/runs.defs.ts | 2 +- web/lib/runs/reprocessable-statuses.ts | 25 ++++ web/lib/runs/row-actions.ts | 4 +- web/tests/integration/files.test.ts | 107 +++++++++++++++++- web/tests/unit/reprocessable-statuses.test.ts | 39 +++++++ 14 files changed, 202 insertions(+), 41 deletions(-) create mode 100644 web/tests/unit/reprocessable-statuses.test.ts diff --git a/developer-docs/lambda.md b/developer-docs/lambda.md index 79d7c73..8572b71 100644 --- a/developer-docs/lambda.md +++ b/developer-docs/lambda.md @@ -18,9 +18,9 @@ Slack notifications are sent by the **web app**, not the Lambda — see [Slack n ### Function URL (manual reprocessing) -When a file fails processing (or needs to be re-run), users can trigger reprocessing from the run detail page in the web app. This invokes the Lambda's Function URL instead of going through S3: +When a raw file fails processing (or needs to be re-run), users can trigger reprocessing from the run detail page in the web app. Processed artifacts (preview CSVs, posters, MP4s) are not eligible — they often share extensions with the watcher's raw patterns, and Lambda only ingests the original instrument files. This invokes the Lambda's Function URL instead of going through S3: -1. The user clicks **Reprocess** on an uploaded, failed, or completed file in the web dashboard. +1. The user clicks **Reprocess** on an uploaded, failed, or completed raw file in the web dashboard, or **Reprocess** on the run (which queues every eligible raw file). 2. The web app's `POST /api/v1/files/:fileId/reprocess` endpoint transitions the file to `processing` status, clears any previous error, and sends a POST request to the Lambda Function URL. 3. The Function URL is configured with `AuthType: AWS_IAM`, so the web app SigV4-signs the request using credentials it gets via Vercel OIDC federation (the `WebAppS3Role` IAM role, which has `lambda:InvokeFunctionUrl` on this function's ARN). The body is a JSON payload containing a synthetic S3 event. 4. The Lambda handler detects the Function URL invocation (via `requestContext` in the event) and parses the S3 event from the request body. Inbound auth is enforced by AWS itself in front of the function — the handler never sees an unauthenticated request. diff --git a/web/app/api/v1/files/[fileId]/reprocess/route.ts b/web/app/api/v1/files/[fileId]/reprocess/route.ts index 79d619f..70dec80 100644 --- a/web/app/api/v1/files/[fileId]/reprocess/route.ts +++ b/web/app/api/v1/files/[fileId]/reprocess/route.ts @@ -14,9 +14,9 @@ interface RouteContext { // --------------------------------------------------------------------------- // POST /api/v1/files/:fileId/reprocess // -// Transitions an uploaded, failed, or completed file back to "processing" -// and invokes the Lambda Function URL to re-run the instrument's -// process_file workflow. +// Transitions an uploaded, failed, or completed raw file back to +// "processing" and invokes the Lambda Function URL to re-run the +// instrument's process_file workflow. Processed artifacts are rejected. // The core logic lives in lib/api/file-reprocessing.ts so the MCP server // can reuse it. // --------------------------------------------------------------------------- diff --git a/web/app/api/v1/instruments/[instrumentId]/runs/[runId]/reprocess/route.ts b/web/app/api/v1/instruments/[instrumentId]/runs/[runId]/reprocess/route.ts index 15876c1..d8f21bd 100644 --- a/web/app/api/v1/instruments/[instrumentId]/runs/[runId]/reprocess/route.ts +++ b/web/app/api/v1/instruments/[instrumentId]/runs/[runId]/reprocess/route.ts @@ -11,8 +11,8 @@ interface RouteContext { // POST /api/v1/instruments/:instrumentId/runs/:runId/reprocess // // Run-level convenience endpoint that reprocesses every `uploaded`, -// `completed`, or `failed` file on the run. Used by the runs list -// row/bulk actions. +// `completed`, or `failed` raw file on the run. Processed artifacts +// are skipped. Used by the runs list row/bulk actions. // --------------------------------------------------------------------------- export async function POST(request: NextRequest, { params }: RouteContext) { diff --git a/web/components/runs/file-selection-provider.tsx b/web/components/runs/file-selection-provider.tsx index ee513fc..c2cb488 100644 --- a/web/components/runs/file-selection-provider.tsx +++ b/web/components/runs/file-selection-provider.tsx @@ -3,8 +3,7 @@ import { createContext, use, useCallback, useMemo, useState } from "react"; import type { RunFile } from "@/lib/api/instrument-runs"; import type { InstrumentType } from "@/lib/db/schema"; -import { isProcessableInstrumentType } from "@/lib/instruments/processable-types"; -import { REPROCESSABLE_STATUSES } from "@/lib/runs/reprocessable-statuses"; +import { canReprocessFile } from "@/lib/runs/reprocessable-statuses"; // --------------------------------------------------------------------------- // File selection provider for the run files table. Mirrors RunSelectionProvider @@ -21,8 +20,6 @@ const DOWNLOADABLE_STATUSES = new Set([ "failed", ]); -const REPROCESSABLE_STATUS_SET = new Set(REPROCESSABLE_STATUSES); - export interface FileCaps { dismiss: boolean; download: boolean; @@ -48,10 +45,7 @@ export function buildFileRef( } const isDetected = file.status === "detected"; const canDownload = DOWNLOADABLE_STATUSES.has(file.status); - const canReprocess = - isProcessableInstrumentType(instrumentType) && - REPROCESSABLE_STATUS_SET.has(file.status) && - file.s3Key !== null; + const canReprocess = canReprocessFile(file, instrumentType); if (!(isDetected || canDownload)) { return null; } diff --git a/web/components/runs/run-files-table.tsx b/web/components/runs/run-files-table.tsx index f87d1e4..1e3de56 100644 --- a/web/components/runs/run-files-table.tsx +++ b/web/components/runs/run-files-table.tsx @@ -25,8 +25,7 @@ import { import type { RunFile } from "@/lib/api/instrument-runs"; import { formatDateTime } from "@/lib/date"; import type { InstrumentType } from "@/lib/db/schema"; -import { isProcessableInstrumentType } from "@/lib/instruments/processable-types"; -import { REPROCESSABLE_STATUSES } from "@/lib/runs/reprocessable-statuses"; +import { canReprocessFile } from "@/lib/runs/reprocessable-statuses"; import { cn, formatBytes } from "@/lib/utils"; import { FileSelectAllCheckbox, @@ -44,8 +43,6 @@ const DOWNLOADABLE_STATUSES = new Set([ "failed", ]); -const REPROCESSABLE_STATUS_SET = new Set(REPROCESSABLE_STATUSES); - const CATEGORY_BADGE_CLASSES: Record = { raw: "bg-blue-100 text-blue-700 dark:bg-blue-950 dark:text-blue-300", processed: @@ -215,19 +212,10 @@ function UploadDismissActions({ ); } -function canReprocess(file: RunFile, instrumentType: InstrumentType): boolean { - return ( - file.deletedAt === null && - isProcessableInstrumentType(instrumentType) && - REPROCESSABLE_STATUS_SET.has(file.status) && - file.s3Key !== null - ); -} - // --------------------------------------------------------------------------- // Read-only variant: no selection column, no upload/dismiss. Reprocessing is -// still allowed for uploaded/completed/failed files so operators can recover -// report data or kick stuck uploads without restoring the run. +// still allowed for uploaded/completed/failed raw files so operators can +// recover report data or kick stuck uploads without restoring the run. // --------------------------------------------------------------------------- export interface ReadOnlyRunFilesTableProps { @@ -261,7 +249,7 @@ export function ReadOnlyRunFilesTable({ > - {canReprocess(file, instrumentType) ? ( + {canReprocessFile(file, instrumentType) ? (
{ }; } + // Processed artifacts often share watcher extensions. Lambda only + // ingests raw inputs; queuing them marks them processing then strands + // them when the processor ignores the filename. + if (file.category !== "raw") { + return { + ok: false, + status: 409, + code: "CONFLICT", + message: + "Cannot reprocess a processed artifact — only raw files can be reprocessed", + }; + } + if (!(REPROCESSABLE_STATUSES as readonly string[]).includes(file.status)) { return { ok: false, @@ -226,6 +239,7 @@ export async function reprocessRun( .where( and( eq(files.instrumentRunId, run.id), + eq(files.category, "raw"), inArray(files.status, [...REPROCESSABLE_STATUSES]), isNull(files.deletedAt) ) diff --git a/web/lib/api/openapi/paths/files.ts b/web/lib/api/openapi/paths/files.ts index 4fe4439..b469f31 100644 --- a/web/lib/api/openapi/paths/files.ts +++ b/web/lib/api/openapi/paths/files.ts @@ -86,7 +86,7 @@ registry.registerPath({ operationId: "reprocessFile", summary: "Reprocess a file", description: - "Requires scope `files:reprocess`. Eligible statuses: `uploaded`, `failed`, or `completed` (file must have an S3 location). The file's instrument must have a Lambda processor.", + "Requires scope `files:reprocess`. Raw files only. Eligible statuses: `uploaded`, `failed`, or `completed` (file must have an S3 location). The file's instrument must have a Lambda processor.", tags: ["Files"], security: bearerSecurity, request: { params: fileParams }, diff --git a/web/lib/api/openapi/paths/runs.ts b/web/lib/api/openapi/paths/runs.ts index cbeaef0..5b62afd 100644 --- a/web/lib/api/openapi/paths/runs.ts +++ b/web/lib/api/openapi/paths/runs.ts @@ -141,7 +141,7 @@ registry.registerPath({ path: "/instruments/{instrumentId}/runs/{runId}/reprocess", operationId: "reprocessInstrumentRun", summary: "Reprocess a run", - description: `${scoped("runs:reprocess")} Reprocesses every \`uploaded\`, \`failed\`, or \`completed\` file on the run. The instrument must have a Lambda processor.`, + description: `${scoped("runs:reprocess")} Reprocesses every \`uploaded\`, \`failed\`, or \`completed\` raw file on the run. Processed artifacts are skipped. The instrument must have a Lambda processor.`, tags: tag, security: bearerSecurity, request: { params: runParams }, diff --git a/web/lib/mcp/tools/files.defs.ts b/web/lib/mcp/tools/files.defs.ts index d8f6ddc..708e01d 100644 --- a/web/lib/mcp/tools/files.defs.ts +++ b/web/lib/mcp/tools/files.defs.ts @@ -51,7 +51,7 @@ export const reprocessFileTool = { name: "reprocess_file", title: "Reprocess File", description: - "Re-run the Lambda processing workflow for an uploaded, failed, or completed file on an instrument that has a Lambda processor. Transitions the file back to 'processing'. Use this to retry after a parser fix, transient Lambda failure, or a stuck upload that never entered processing.", + "Re-run the Lambda processing workflow for an uploaded, failed, or completed raw file on an instrument that has a Lambda processor. Processed artifacts are rejected. Transitions the file back to 'processing'. Use this to retry after a parser fix, transient Lambda failure, or a stuck upload that never entered processing.", group: "files", scope: "files:reprocess", inputSchema: { fileId: z.number().int().describe("Numeric file ID") }, diff --git a/web/lib/mcp/tools/runs.defs.ts b/web/lib/mcp/tools/runs.defs.ts index 713e5fb..b3cb291 100644 --- a/web/lib/mcp/tools/runs.defs.ts +++ b/web/lib/mcp/tools/runs.defs.ts @@ -317,7 +317,7 @@ export const reprocessRunTool = { scope: "runs:reprocess", title: "Reprocess Run", description: - "Re-run Lambda processing for every uploaded, completed, or failed file on a run. The instrument must have a Lambda processor. Prefer this over looping reprocess_file for bulk retries after a parser fix or to kick stuck uploads.", + "Re-run Lambda processing for every uploaded, completed, or failed raw file on a run. Processed artifacts are skipped. The instrument must have a Lambda processor. Prefer this over looping reprocess_file for bulk retries after a parser fix or to kick stuck uploads.", inputSchema: { instrumentId: z.string().describe("Instrument identifier"), runId: z.string().describe("Run identifier within the instrument"), diff --git a/web/lib/runs/reprocessable-statuses.ts b/web/lib/runs/reprocessable-statuses.ts index 8c4b96d..ec676ca 100644 --- a/web/lib/runs/reprocessable-statuses.ts +++ b/web/lib/runs/reprocessable-statuses.ts @@ -1,3 +1,6 @@ +import type { InstrumentType } from "@/lib/db/schema"; +import { isProcessableInstrumentType } from "@/lib/instruments/processable-types"; + // Statuses eligible for POST /files/:id/reprocess (and run-level reprocess). // Includes `uploaded` so stuck S3 uploads can be kicked when the Lambda // trigger never fired. Client-safe — imported by UI tables and the API. @@ -6,3 +9,25 @@ export const REPROCESSABLE_STATUSES = [ "failed", "completed", ] as const; + +const REPROCESSABLE_STATUS_SET = new Set(REPROCESSABLE_STATUSES); + +interface ReprocessableFile { + category: string; + deletedAt: Date | string | null; + s3Key: string | null; + status: string; +} + +export function canReprocessFile( + file: ReprocessableFile, + instrumentType: InstrumentType +): boolean { + return ( + file.category === "raw" && + file.deletedAt === null && + file.s3Key !== null && + REPROCESSABLE_STATUS_SET.has(file.status) && + isProcessableInstrumentType(instrumentType) + ); +} diff --git a/web/lib/runs/row-actions.ts b/web/lib/runs/row-actions.ts index 8d44aae..adb8c83 100644 --- a/web/lib/runs/row-actions.ts +++ b/web/lib/runs/row-actions.ts @@ -14,8 +14,8 @@ import { isProcessableInstrumentType } from "@/lib/instruments/processable-types // - Upload: at least one file still waiting to be uploaded. // - Download: at least one file has made it to S3 (i.e. is not `detected` // or `upload_requested`). -// - Reprocess: instrument_type has a Lambda processor and at least one file -// is in `uploaded`, `completed`, or `failed`. +// - Reprocess: instrument_type has a Lambda processor and at least one raw +// file is in `uploaded`, `completed`, or `failed`. // - Delete: the run isn't already soft-deleted. // --------------------------------------------------------------------------- diff --git a/web/tests/integration/files.test.ts b/web/tests/integration/files.test.ts index 140f56c..64d8fd4 100644 --- a/web/tests/integration/files.test.ts +++ b/web/tests/integration/files.test.ts @@ -1,6 +1,6 @@ import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { fileDetail, fileDismissed } from "@/lib/api/openapi"; +import { fileDetail, fileDismissed, runReprocessed } from "@/lib/api/openapi"; import { instruments, files as schemaFiles } from "@/lib/db/schema"; import { api, @@ -489,7 +489,7 @@ describe("Files API", () => { // fileId (sample.csv) → completed, has S3 info // secondFileId (sample2.csv) → detected, soft-deleted // thirdFileId (sample3.csv) → detected, not deleted - // lambdaFileId (processed_output) → failed, has S3 info + // lambdaFileId (processed_output) → failed processed artifact, has S3 info // Uploaded eligibility is covered by a dedicated run/file created below. // ------------------------------------------------------------------------- @@ -587,11 +587,52 @@ describe("Files API", () => { expect(data.error.message).toContain("parent run"); }); + it("REPROCESS returns 409 for a processed artifact", async () => { + const res = await api(`/api/v1/files/${lambdaFileId}/reprocess`, { + method: "POST", + token, + }); + expect(res.status).toBe(409); + const data = await res.json(); + expect(data.error.message).toContain("processed"); + }); + // These tests verify that reprocessable statuses (uploaded, failed, and // completed) pass all validation guards. They return 503 because the // test server has no LAMBDA_FUNCTION_URL configured. it("REPROCESS returns 503 for failed file when Lambda is not configured", async () => { - const res = await api(`/api/v1/files/${lambdaFileId}/reprocess`, { + const failedRunId = "reprocess-failed-raw-run"; + await api(`/api/v1/instruments/${instrumentId}/runs`, { + method: "POST", + token, + body: { run_id: failedRunId, source: "lambda" }, + }); + const createFileRes = await api( + `/api/v1/instruments/${instrumentId}/runs/${failedRunId}/files`, + { + method: "POST", + token, + body: { + s3_bucket: "test-bucket", + s3_key: `${instrumentId}/${failedRunId}/raw.csv`, + filename: "raw.csv", + }, + } + ); + expect(createFileRes.status).toBe(201); + const createdFile = await createFileRes.json(); + await api(`/api/v1/files/${createdFile.id}`, { + method: "PATCH", + token, + body: { status: "processing" }, + }); + await api(`/api/v1/files/${createdFile.id}`, { + method: "PATCH", + token, + body: { status: "failed", error_message: "parser error" }, + }); + + const res = await api(`/api/v1/files/${createdFile.id}/reprocess`, { method: "POST", token, }); @@ -722,4 +763,64 @@ describe("Files API", () => { expect(data.error.message).toContain("generic"); expect(data.error.message).toContain("no Lambda processor"); }); + + // ------------------------------------------------------------------------- + // POST /api/v1/instruments/:instrumentId/runs/:runId/reprocess + // ------------------------------------------------------------------------- + + it("RUN REPROCESS queues only raw files", async () => { + const skipRunId = "reprocess-skip-processed-run"; + await api(`/api/v1/instruments/${instrumentId}/runs`, { + method: "POST", + token, + body: { run_id: skipRunId, source: "lambda" }, + }); + const rawRes = await api( + `/api/v1/instruments/${instrumentId}/runs/${skipRunId}/files`, + { + method: "POST", + token, + body: { + s3_bucket: "test-bucket", + s3_key: `${instrumentId}/${skipRunId}/stack.tif`, + filename: "stack.tif", + }, + } + ); + const processedRes = await api( + `/api/v1/instruments/${instrumentId}/runs/${skipRunId}/files`, + { + method: "POST", + token, + body: { + s3_bucket: "processed-bucket", + s3_key: `${instrumentId}/${skipRunId}/stack.jpg`, + filename: "stack.jpg", + category: "processed", + }, + } + ); + expect(rawRes.status).toBe(201); + expect(processedRes.status).toBe(201); + const processedFile = await processedRes.json(); + + const res = await api( + `/api/v1/instruments/${instrumentId}/runs/${skipRunId}/reprocess`, + { method: "POST", token } + ); + expect(res.status).toBe(200); + const data = await res.json(); + runReprocessed.parse(data); + // Lambda is not configured, so the raw file is attempted and fails. + // The processed poster must not be in that set. + expect(data.files_queued).toBe(0); + expect(data.files_failed).toBe(1); + + const db = getTestDb(); + const [processed] = await db + .select({ status: schemaFiles.status }) + .from(schemaFiles) + .where(eq(schemaFiles.id, processedFile.id)); + expect(processed?.status).toBe("uploaded"); + }); }); diff --git a/web/tests/unit/reprocessable-statuses.test.ts b/web/tests/unit/reprocessable-statuses.test.ts new file mode 100644 index 0000000..af09c45 --- /dev/null +++ b/web/tests/unit/reprocessable-statuses.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from "vitest"; +import { canReprocessFile } from "@/lib/runs/reprocessable-statuses"; + +const rawCompleted = { + category: "raw", + deletedAt: null, + s3Key: "dishcam/run/stack.tif", + status: "completed", +}; + +describe("canReprocessFile", () => { + it("allows a completed raw file on a processable instrument", () => { + expect(canReprocessFile(rawCompleted, "dishcam")).toBe(true); + }); + + it("rejects processed artifacts even when status and S3 look eligible", () => { + expect( + canReprocessFile( + { + ...rawCompleted, + category: "processed", + s3Key: "dishcam/run/stack.jpg", + status: "uploaded", + }, + "dishcam" + ) + ).toBe(false); + }); + + it("rejects detected files, missing S3, and unprocessable types", () => { + expect( + canReprocessFile({ ...rawCompleted, status: "detected" }, "dishcam") + ).toBe(false); + expect(canReprocessFile({ ...rawCompleted, s3Key: null }, "dishcam")).toBe( + false + ); + expect(canReprocessFile(rawCompleted, "generic")).toBe(false); + }); +}); From e8a74c88152e02c7b3eec8ad1ef681bfc90beda0 Mon Sep 17 00:00:00 2001 From: Wasim Amiri <7220175+wasimxyz@users.noreply.github.com> Date: Fri, 21 Aug 2026 09:39:23 -0700 Subject: [PATCH 2/2] Fail a reprocess the processor can't handle instead of stranding it Skipping the filename gate on Function URL invokes was meant to keep a user-initiated reprocess from being silently dropped, but processors return without touching the file row when the name isn't theirs, so the web app's `processing` transition was left with nothing to resolve it. The gate now always runs and a mismatch is PATCHed to `failed`, matching the existing recovery for a missing instrument or unmapped type. Also derive `ReprocessableFile` from the `files` row so a mistyped category or status literal fails to compile. Co-authored-by: Cursor --- developer-docs/lambda.md | 5 +- lambda/src/data_hub_lambda/handler.py | 21 +++++--- lambda/tests/test_processors.py | 52 +++++++++++++++++-- web/lib/runs/reprocessable-statuses.ts | 14 ++--- web/tests/unit/reprocessable-statuses.test.ts | 7 ++- 5 files changed, 78 insertions(+), 21 deletions(-) diff --git a/developer-docs/lambda.md b/developer-docs/lambda.md index 8572b71..c55eca7 100644 --- a/developer-docs/lambda.md +++ b/developer-docs/lambda.md @@ -24,7 +24,8 @@ When a raw file fails processing (or needs to be re-run), users can trigger repr 2. The web app's `POST /api/v1/files/:fileId/reprocess` endpoint transitions the file to `processing` status, clears any previous error, and sends a POST request to the Lambda Function URL. 3. The Function URL is configured with `AuthType: AWS_IAM`, so the web app SigV4-signs the request using credentials it gets via Vercel OIDC federation (the `WebAppS3Role` IAM role, which has `lambda:InvokeFunctionUrl` on this function's ARN). The body is a JSON payload containing a synthetic S3 event. 4. The Lambda handler detects the Function URL invocation (via `requestContext` in the event) and parses the S3 event from the request body. Inbound auth is enforced by AWS itself in front of the function — the handler never sees an unauthenticated request. -5. From here, processing follows the same type-based dispatch as the S3 trigger path, except **filename gates are skipped** — a user clicking Reprocess has stated intent, so the handler must not leave the file stranded in `processing`. +5. From here, processing follows the same type-based dispatch as the S3 trigger path, except the cheap **union gate is skipped** so the handler always reaches the instrument lookup. +6. If the matched processor's own filename gate rejects the file, the handler PATCHes it to `failed` with `The {instrument_type} processor does not handle '{filename}'` instead of returning quietly. The web app moved the row to `processing` before invoking, and processors return without touching the row when the name isn't theirs — so a silent no-op would strand it there forever. The same recovery covers a missing instrument and an unmapped `instrument_type`. ### Function URL (archive build) @@ -41,7 +42,7 @@ See [Run archives](run-archives.md) for the full flow, S3 bucket layout, cache s Dispatch is by `instrument_type` (Postgres/TS enum), not instrument ID. The registry lives in `lambda/src/data_hub_lambda/processors.py`; the web reprocess gate mirrors the same keys in `web/lib/instruments/processable-types.ts`. -| `instrument_type` | Module | Filename gate (S3 events only) | +| `instrument_type` | Module | Filename gate | | --- | --- | --- | | `tape_station` | `agilent_4150_tapestation` | `.pdf` | | `fplc` | `akta_fplc` | `.pdf` | diff --git a/lambda/src/data_hub_lambda/handler.py b/lambda/src/data_hub_lambda/handler.py index 465bd87..4ebfe5b 100644 --- a/lambda/src/data_hub_lambda/handler.py +++ b/lambda/src/data_hub_lambda/handler.py @@ -309,15 +309,16 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | """Top-level Lambda handler dispatching to instrument workflows. Dispatch is by ``instrument_type`` (fetched from the API), not instrument - ID. Filename gates filter the S3 firehose; explicit reprocess via the - Function URL bypasses those gates so user-initiated work can't strand - a file in ``processing``. + ID. Filename gates filter the S3 firehose, which delivers every upload. + A reprocess via the Function URL skips the cheap union gate so it always + reaches the instrument lookup, then fails the file if the processor's own + gate rejects it — the web app already moved the row to ``processing``, so + a silent return would strand it. """ logger.info("Received event: %s", pformat(event)) - # Capture before unwrapping so reprocess (Function URL) can skip gates. + # Capture before unwrapping: the payload loses the Function URL envelope. is_function_url = _is_function_url_event(event) - apply_filename_gates = not is_function_url # Function URL invocations carry a requestContext with an http key. # AWS_IAM auth is enforced by Lambda before the handler runs, so we @@ -354,7 +355,8 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | try: # Skip the API call when no processor could possibly want this filename. - if apply_filename_gates and not matches_any_processor_gate(filename): + # Reprocess skips this so it can name the instrument type in the error. + if not is_function_url and not matches_any_processor_gate(filename): logger.info( "No processor filename gate matches %s; skipping.", filename, @@ -410,12 +412,17 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | _fail_reprocess_file(instrument_id, run_id, filename, message) return None - if apply_filename_gates and not processor.matches_filename(filename): + if not processor.matches_filename(filename): + message = f"The {instrument.instrument_type} processor does not handle '{filename}'" logger.info( "Filename %s does not match gate for instrument_type=%s; skipping.", filename, instrument.instrument_type, ) + # Processors return without touching the file row when the name + # isn't theirs, so a reprocess has to report the mismatch itself. + if is_function_url: + _fail_reprocess_file(instrument_id, run_id, filename, message) return None logger.info( diff --git a/lambda/tests/test_processors.py b/lambda/tests/test_processors.py index 3d286f6..c6165de 100644 --- a/lambda/tests/test_processors.py +++ b/lambda/tests/test_processors.py @@ -340,7 +340,8 @@ def test_403_raises(self) -> None: assert exc_info.value.status_code == 403 - def test_reprocess_bypasses_filename_gate(self) -> None: + def test_reprocess_marks_failed_when_filename_unsupported(self) -> None: + """A reprocess the processor would ignore must not stay in `processing`.""" client = MagicMock() client.get_instrument.return_value = InstrumentResponse( id="azure-cielo-qpcr", @@ -348,6 +349,7 @@ def test_reprocess_bypasses_filename_gate(self) -> None: status="active", instrument_type="qpcr", ) + client.create_file.return_value = self._file_response(11) process_file = MagicMock() entry = MagicMock() entry.matches_filename.return_value = False @@ -360,11 +362,16 @@ def test_reprocess_bypasses_filename_gate(self) -> None: "data_hub_lambda.handler.get_processor", return_value=entry, ), - # Union gate would also block S3 events; reprocess must skip it. + # Union gate would short-circuit an S3 event before the API call; + # reprocess must skip it so the error can name the type. patch( "data_hub_lambda.handler.matches_any_processor_gate", return_value=False, ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "test-bucket", + ), ): result = lambda_handler( self._function_url_event( @@ -375,12 +382,51 @@ def test_reprocess_bypasses_filename_gate(self) -> None: MagicMock(), ) + assert result is None + client.get_instrument.assert_called_once_with("azure-cielo-qpcr") + process_file.assert_not_called() + client.update_file.assert_called_once_with( + 11, + status="failed", + error_message=( + "The qpcr processor does not handle 'Experiment_Amplification Results.csv'" + ), + ) + + def test_reprocess_runs_when_filename_matches(self) -> None: + client = MagicMock() + client.get_instrument.return_value = InstrumentResponse( + id="azure-cielo-qpcr", + display_name="Azure Cielo qPCR", + status="active", + instrument_type="qpcr", + ) + process_file = MagicMock() + entry = MagicMock() + entry.matches_filename.return_value = True + entry.process_file = process_file + + with ( + patch("data_hub_lambda.handler.get_client", return_value=client), + patch("data_hub_lambda.handler._cleanup_tmp"), + patch("data_hub_lambda.handler.get_processor", return_value=entry), + ): + result = lambda_handler( + self._function_url_event( + "azure-cielo-qpcr", + "run-1", + "Experiment_Cq Values.csv", + ), + MagicMock(), + ) + assert result is None process_file.assert_called_once_with( "azure-cielo-qpcr", "run-1", - "Experiment_Amplification Results.csv", + "Experiment_Cq Values.csv", ) + client.update_file.assert_not_called() def test_s3_event_applies_per_type_gate(self) -> None: client = MagicMock() diff --git a/web/lib/runs/reprocessable-statuses.ts b/web/lib/runs/reprocessable-statuses.ts index ec676ca..df8900c 100644 --- a/web/lib/runs/reprocessable-statuses.ts +++ b/web/lib/runs/reprocessable-statuses.ts @@ -1,4 +1,4 @@ -import type { InstrumentType } from "@/lib/db/schema"; +import type { files, InstrumentType } from "@/lib/db/schema"; import { isProcessableInstrumentType } from "@/lib/instruments/processable-types"; // Statuses eligible for POST /files/:id/reprocess (and run-level reprocess). @@ -12,12 +12,12 @@ export const REPROCESSABLE_STATUSES = [ const REPROCESSABLE_STATUS_SET = new Set(REPROCESSABLE_STATUSES); -interface ReprocessableFile { - category: string; - deletedAt: Date | string | null; - s3Key: string | null; - status: string; -} +// Only the columns the predicate reads, taken from the `files` row so a typo +// in a category or status literal fails to compile. +export type ReprocessableFile = Pick< + typeof files.$inferSelect, + "category" | "deletedAt" | "s3Key" | "status" +>; export function canReprocessFile( file: ReprocessableFile, diff --git a/web/tests/unit/reprocessable-statuses.test.ts b/web/tests/unit/reprocessable-statuses.test.ts index af09c45..2063acd 100644 --- a/web/tests/unit/reprocessable-statuses.test.ts +++ b/web/tests/unit/reprocessable-statuses.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it } from "vitest"; -import { canReprocessFile } from "@/lib/runs/reprocessable-statuses"; +import { + canReprocessFile, + type ReprocessableFile, +} from "@/lib/runs/reprocessable-statuses"; -const rawCompleted = { +const rawCompleted: ReprocessableFile = { category: "raw", deletedAt: null, s3Key: "dishcam/run/stack.tif",