From a8dffc338168db27f1fb527a3352b1437761817c Mon Sep 17 00:00:00 2001 From: Dhyey Desai Date: Fri, 2 Oct 2026 21:02:40 -0400 Subject: [PATCH] fix: mark runs failed when the SQS send fails instead of leaving them queued approve-rules and auto-validate move a run to 'queued' and then send to the executor queue; profile-complete moves it to 'awaiting_ai' and sends to the AI jobs queue. When SendMessage failed, approve-rules and auto-validate only logged and returned 200, and profile-complete threw a bare 500. Either way the run sat in its in-between status with no message until reconcile-runs failed it 20 minutes later. Now each route marks the run failed with a message the run page shows and returns 503 { error, run_status: 'failed' }. The shared helper (src/lib/queue-failure.ts) updates only if the run is still in the status the route set, because SendMessage can fail after SQS accepted the message and a worker may already have claimed the run. suggest-transforms no longer overwrites that message with 'auto-validate error 503: ...'. Tests: unit tests for all three routes, and an integration test that sends to a LocalStack queue that does not exist. --- README.md | 2 +- docs/operations.md | 3 +- src/app/api/approve-rules/route.test.ts | 24 +++++++ src/app/api/approve-rules/route.ts | 10 +-- .../api/auto-validate/[runId]/route.test.ts | 69 +++++++++++++++++++ src/app/api/auto-validate/[runId]/route.ts | 8 ++- src/app/api/suggest-transforms/route.ts | 6 +- .../webhooks/profile-complete/route.test.ts | 54 +++++++++++++++ .../api/webhooks/profile-complete/route.ts | 19 +++-- src/lib/queue-failure.ts | 48 +++++++++++++ tests/integration/approve-rules-claim.test.ts | 19 +++++ 11 files changed, 247 insertions(+), 15 deletions(-) create mode 100644 src/app/api/auto-validate/[runId]/route.test.ts create mode 100644 src/app/api/webhooks/profile-complete/route.test.ts create mode 100644 src/lib/queue-failure.ts diff --git a/README.md b/README.md index 146ad58..85a363e 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,7 @@ Operations runbook: [docs/operations.md](docs/operations.md). Security model: [d - **No Row-Level Security.** Tenant isolation depends on every query filtering on `team_id`. - **No infrastructure as code.** Buckets, queues, IAM, Lambda settings and schedules cannot be reviewed or reproduced from this repo. -- **No automatic re-enqueue.** If the SQS send after approval fails, the run stays `queued` until the reconciler marks it failed ([runbook](docs/operations.md)). +- **No automatic re-enqueue.** If sending to the executor or AI jobs queue fails, the run is marked failed right away and the user has to upload the file again ([runbook](docs/operations.md)). - **Retention:** raw files of failed runs stay in S3, and `data_retention_days` is not enforced. The `guest_` lifecycle rule is the only exception. - **The AI budget is checked before a call and recorded after it.** Concurrent requests can overshoot a cap by the cost of the calls in flight. Costs are estimates, not the AWS bill. - **Upload limits are per file.** A file under the limit can still exhaust the executor's memory or time on expensive rules. Only `semantic_deduplicate` has a time and row guard. diff --git a/docs/operations.md b/docs/operations.md index be2bf81..0b6eca1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -4,7 +4,8 @@ | Symptom | What to check or do | |---|---| -| Run stuck in `profiling`, `awaiting_ai`, `queued` or `running` | `reconcile-runs` marks these runs `failed` after 20 minutes, so first confirm that the EventBridge rule `cleanstack-reconciler-5min` is firing. For a run stuck in `queued`, search the log for `SQS send failed; run left queued without a message`. If you find it, re-send the message: `aws sqs send-message --queue-url "$SQS_QUEUE_URL" --message-body '{"run_id":""}'`. This is safe because execution is idempotent. | +| Run stuck in `profiling`, `awaiting_ai`, `queued` or `running` | `reconcile-runs` marks these runs `failed` after 20 minutes, so first confirm that the EventBridge rule `cleanstack-reconciler-5min` is firing. | +| Run `failed` with "Could not queue this run …" | Sending to SQS failed (log `SQS send failed; marking run failed` or `AI jobs SQS send failed; marking run failed`), so the route failed the run and returned 503. Check the queue URL and the `sqs:SendMessage` permission of `cleanstack-vercel`. The user retries by uploading again. | | Runs stay in `pending` after upload | The profiler trigger is missing `Post`. It must fire on all object create events. | | Run `failed` with `Attempt n/3 failed, retrying` in its history | SQS retries transient errors up to `EXECUTOR_MAX_ATTEMPTS` times, and the last attempt marks the run failed. Look up the run id in the executor's CloudWatch logs. | | Approved rule shows **Not applied** | This is expected when a rule is unsafe or invalid. The page shows the reason, and the rest of the run is unaffected. | diff --git a/src/app/api/approve-rules/route.test.ts b/src/app/api/approve-rules/route.test.ts index b033e63..0118aaf 100644 --- a/src/app/api/approve-rules/route.test.ts +++ b/src/app/api/approve-rules/route.test.ts @@ -61,6 +61,30 @@ describe("POST /api/approve-rules", () => { expect(sqsSend).not.toHaveBeenCalled(); }); + it("marks the run failed and returns 503 when the SQS send fails", async () => { + queryOne.mockImplementation(async (sql: string) => + sql.includes("RETURNING id") ? { id: RUN } : null); + sqsSend.mockRejectedValueOnce(new Error("AWS.SimpleQueueService.NonExistentQueue")); + const res = await post({ run_id: RUN, rule_decisions: decisions }); + expect(res.status).toBe(503); + const body = await res.json(); + expect(body.run_status).toBe("failed"); + expect(body.error).toMatch(/Could not queue this run for execution/); + const fail = queryOne.mock.calls.find((c) => String(c[0]).includes("SET status = 'failed'"))!; + expect(String(fail[0])).toContain("WHERE id = $1 AND status = $2"); + expect(fail[1]).toEqual([RUN, "queued", body.error]); + }); + + it("still returns 503 when marking the run failed also fails", async () => { + queryOne.mockImplementation(async (sql: string) => { + if (sql.includes("SET status = 'failed'")) throw new Error("db down"); + return sql.includes("RETURNING id") ? { id: RUN } : null; + }); + sqsSend.mockRejectedValueOnce(new Error("timeout")); + const res = await post({ run_id: RUN, rule_decisions: decisions }); + expect(res.status).toBe(503); + }); + it("validates decisions", async () => { expect((await post({ run_id: RUN, rule_decisions: [{ rule_id: 1, action: "maybe" }] })).status).toBe(400); }); diff --git a/src/app/api/approve-rules/route.ts b/src/app/api/approve-rules/route.ts index d21d233..12fc392 100644 --- a/src/app/api/approve-rules/route.ts +++ b/src/app/api/approve-rules/route.ts @@ -4,6 +4,7 @@ import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs"; import { query, queryOne, queryOneWithTeam, withTransaction } from "@/lib/db"; import { requireEnv, optionalEnv, awsRegion } from "@/lib/env"; import { logger } from "@/lib/logger"; +import { EXECUTOR_QUEUE_FAILED_MESSAGE, failRunAfterQueueError, queueErrorResponse } from "@/lib/queue-failure"; const log = logger.child({ route: "POST /api/approve-rules" }); @@ -108,10 +109,11 @@ export async function POST(req: NextRequest) { }) ); } catch (sqsErr) { - // The run stays 'queued' with no message. Nothing re-enqueues it: the - // reconcile-runs cron only marks stale runs 'failed' (see README "Known - // limitations"). An operator can re-send {"run_id": ...} to the queue. - log.error("SQS send failed; run left queued without a message", { run_id, err: sqsErr }); + // Without a message the run would sit in 'queued' until reconcile-runs + // fails it 20 minutes later, so fail it now (src/lib/queue-failure.ts). + log.error("SQS send failed; marking run failed", { run_id, err: sqsErr }); + await failRunAfterQueueError(run_id, "queued", EXECUTOR_QUEUE_FAILED_MESSAGE); + return queueErrorResponse(EXECUTOR_QUEUE_FAILED_MESSAGE); } } diff --git a/src/app/api/auto-validate/[runId]/route.test.ts b/src/app/api/auto-validate/[runId]/route.test.ts new file mode 100644 index 0000000..bbe3e18 --- /dev/null +++ b/src/app/api/auto-validate/[runId]/route.test.ts @@ -0,0 +1,69 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +const query = vi.fn(); +const queryOne = vi.fn(); +const sqsSend = vi.fn(); +const generateText = vi.fn(); + +vi.mock("@/lib/db", () => ({ + query: (...a: unknown[]) => query(...a), + queryOne: (...a: unknown[]) => queryOne(...a), +})); +vi.mock("@aws-sdk/client-sqs", () => ({ + SQSClient: class { send = (...a: unknown[]) => sqsSend(...a); }, + SendMessageCommand: class { constructor(public input: unknown) {} }, +})); +vi.mock("ai", () => ({ generateText: (...a: unknown[]) => generateText(...a) })); +vi.mock("@/lib/ai-model", () => ({ languageModel: () => ({}) })); +vi.mock("@/lib/bedrock-meter", () => ({ + checkAiBudget: async () => ({ ok: true }), + meterBedrockCall: async () => {}, +})); + +const SECRET = "w".repeat(40); +process.env.WEBHOOK_SECRET = SECRET; +const { POST } = await import("./route"); + +const RUN = "11111111-2222-3333-4444-555555555555"; +const call = () => + POST(new NextRequest(`http://x/api/auto-validate/${RUN}`, { + method: "POST", + headers: { "x-webhook-secret": SECRET }, + }), { params: Promise.resolve({ runId: RUN }) }); + +describe("POST /api/auto-validate/[runId]", () => { + beforeEach(() => { + vi.clearAllMocks(); + process.env.SQS_QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/1/cleanstack-jobs"; + query.mockResolvedValue([{ id: "r1", rule_type: "trim_whitespace", column_name: "name", ai_reasoning: "x" }]); + queryOne.mockImplementation(async (sql: string) => { + if (sql.includes("FROM data_profiles")) return { column_stats: {}, total_rows: 10 }; + if (sql.includes("p.team_id")) return { team_id: "user_1", pipeline_id: "p" }; + if (sql.includes("RETURNING id")) return { id: RUN }; + return null; + }); + generateText.mockResolvedValue({ + text: JSON.stringify({ votes: [{ rule_id: "r1", vote: "APPROVE", reason: "ok" }] }), + usage: { inputTokens: 1, outputTokens: 1 }, + }); + }); + + it("queues the run and enqueues it once", async () => { + const res = await call(); + expect(res.status).toBe(200); + expect(sqsSend).toHaveBeenCalledTimes(1); + expect(queryOne.mock.calls.some((c) => String(c[0]).includes("status = 'queued'"))).toBe(true); + }); + + it("marks the run failed and returns 503 when the SQS send fails", async () => { + sqsSend.mockRejectedValueOnce(new Error("NonExistentQueue")); + const res = await call(); + expect(res.status).toBe(503); + const body = await res.json(); + expect(body.run_status).toBe("failed"); + expect(body.error).toMatch(/Could not queue this run for execution/); + const fail = queryOne.mock.calls.find((c) => String(c[0]).includes("SET status = 'failed'"))!; + expect(fail[1]).toEqual([RUN, "queued", body.error]); + }); +}); diff --git a/src/app/api/auto-validate/[runId]/route.ts b/src/app/api/auto-validate/[runId]/route.ts index c23d0eb..a3942a5 100644 --- a/src/app/api/auto-validate/[runId]/route.ts +++ b/src/app/api/auto-validate/[runId]/route.ts @@ -10,6 +10,7 @@ import { checkAiBudget, meterBedrockCall } from "@/lib/bedrock-meter"; import type { TransformRule, DataProfile } from "@/lib/types"; import { requireEnv, optionalEnv, awsRegion } from "@/lib/env"; import { logger } from "@/lib/logger"; +import { EXECUTOR_QUEUE_FAILED_MESSAGE, failRunAfterQueueError, queueErrorResponse } from "@/lib/queue-failure"; const log = logger.child({ route: "POST /api/auto-validate" }); @@ -264,7 +265,6 @@ ${responseFormat}`, ]); if (approved.length > 0 && optionalEnv("SQS_QUEUE_URL")) { - // Set status before SQS so reconciler can pick up the run if SQS fails await queryOne("UPDATE pipeline_runs SET status = 'queued', updated_at = now() WHERE id = $1", [runId]); try { await sqs.send( @@ -274,7 +274,11 @@ ${responseFormat}`, }) ); } catch (sqsErr) { - log.error("SQS send failed; run left queued without a message (no automatic re-enqueue)", { run_id: runId, err: sqsErr }); + // Without a message the run would sit in 'queued' until reconcile-runs + // fails it 20 minutes later, so fail it now (src/lib/queue-failure.ts). + log.error("SQS send failed; marking run failed", { run_id: runId, err: sqsErr }); + await failRunAfterQueueError(runId, "queued", EXECUTOR_QUEUE_FAILED_MESSAGE); + return queueErrorResponse(EXECUTOR_QUEUE_FAILED_MESSAGE); } } else { // No approved rules — mark completed, nothing to execute diff --git a/src/app/api/suggest-transforms/route.ts b/src/app/api/suggest-transforms/route.ts index c8da9f4..6eb4315 100644 --- a/src/app/api/suggest-transforms/route.ts +++ b/src/app/api/suggest-transforms/route.ts @@ -369,7 +369,8 @@ IMPORTANT RULES: if (!avRes.ok) { const body = await avRes.text().catch(() => ""); log.error("auto-validate call failed", { run_id, status: avRes.status, body: body.slice(0, 500) }); - await queryOne("UPDATE pipeline_runs SET status = 'failed', error_message = $2, updated_at = now() WHERE id = $1", + // Keep the message if auto-validate already failed the run (e.g. queue send error). + await queryOne("UPDATE pipeline_runs SET status = 'failed', error_message = $2, updated_at = now() WHERE id = $1 AND status <> 'failed'", [run_id, `auto-validate error ${avRes.status}: ${body.slice(0, 200)}`]); } } else { @@ -716,7 +717,8 @@ For each rule, write ai_reasoning as one precise sentence that references the sp if (!avRes.ok) { const body = await avRes.text().catch(() => ""); log.error("auto-validate call failed", { run_id, status: avRes.status, body: body.slice(0, 500) }); - await queryOne("UPDATE pipeline_runs SET status = 'failed', error_message = $2, updated_at = now() WHERE id = $1", + // Keep the message if auto-validate already failed the run (e.g. queue send error). + await queryOne("UPDATE pipeline_runs SET status = 'failed', error_message = $2, updated_at = now() WHERE id = $1 AND status <> 'failed'", [run_id, `auto-validate error ${avRes.status}: ${body.slice(0, 200)}`]); } } else { diff --git a/src/app/api/webhooks/profile-complete/route.test.ts b/src/app/api/webhooks/profile-complete/route.test.ts new file mode 100644 index 0000000..e2b2ead --- /dev/null +++ b/src/app/api/webhooks/profile-complete/route.test.ts @@ -0,0 +1,54 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +const queryOne = vi.fn(); +const sqsSend = vi.fn(); + +vi.mock("@/lib/db", () => ({ queryOne: (...a: unknown[]) => queryOne(...a) })); +vi.mock("@aws-sdk/client-sqs", () => ({ + SQSClient: class { send = (...a: unknown[]) => sqsSend(...a); }, + SendMessageCommand: class { constructor(public input: unknown) {} }, +})); + +const SECRET = "w".repeat(40); +process.env.WEBHOOK_SECRET = SECRET; +process.env.AI_QUEUE_ENABLED = "true"; // read at module load +process.env.AI_JOBS_QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/1/ai-jobs"; +const { POST } = await import("./route"); + +const RUN = "11111111-2222-3333-4444-555555555555"; +const post = () => + POST(new NextRequest("http://x/api/webhooks/profile-complete", { + method: "POST", + headers: { "x-webhook-secret": SECRET }, + body: JSON.stringify({ run_id: RUN }), + })); + +describe("POST /api/webhooks/profile-complete (AI queue enabled)", () => { + beforeEach(() => { + vi.clearAllMocks(); + queryOne.mockImplementation(async (sql: string) => + sql.startsWith("SELECT") + ? { id: RUN, pipeline_id: "p", raw_s3_key: `user_1/p/${RUN}/raw.csv`, status: "profiling" } + : sql.includes("RETURNING id") ? { id: RUN } : null); + }); + + it("enqueues the AI job", async () => { + const res = await post(); + expect(res.status).toBe(200); + expect(await res.json()).toEqual({ ok: true, queued: true }); + expect(sqsSend).toHaveBeenCalledTimes(1); + expect(queryOne.mock.calls.some((c) => String(c[0]).includes("'failed'"))).toBe(false); + }); + + it("marks the run failed and returns 503 when the SQS send fails", async () => { + sqsSend.mockRejectedValueOnce(new Error("AccessDenied")); + const res = await post(); + expect(res.status).toBe(503); + const body = await res.json(); + expect(body).toMatchObject({ run_status: "failed" }); + expect(body.error).toMatch(/Could not queue this run for AI suggestions/); + const fail = queryOne.mock.calls.find((c) => String(c[0]).includes("SET status = 'failed'"))!; + expect(fail[1]).toEqual([RUN, "awaiting_ai", body.error]); + }); +}); diff --git a/src/app/api/webhooks/profile-complete/route.ts b/src/app/api/webhooks/profile-complete/route.ts index 0c1cbda..c2f22cd 100644 --- a/src/app/api/webhooks/profile-complete/route.ts +++ b/src/app/api/webhooks/profile-complete/route.ts @@ -4,6 +4,7 @@ import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs"; import { queryOne } from "@/lib/db"; import { requireEnv, optionalEnv, awsRegion } from "@/lib/env"; import { logger } from "@/lib/logger"; +import { AI_QUEUE_FAILED_MESSAGE, failRunAfterQueueError, queueErrorResponse } from "@/lib/queue-failure"; const log = logger.child({ route: "POST /api/webhooks/profile-complete" }); @@ -60,11 +61,19 @@ export async function POST(req: NextRequest) { if (AI_QUEUE_ENABLED) { // Async path: enqueue and return immediately — profiler no longer blocks - await sqs.send(new SendMessageCommand({ - QueueUrl: requireEnv("AI_JOBS_QUEUE_URL"), - MessageBody: JSON.stringify({ run_id }), - MessageGroupId: undefined, - })); + try { + await sqs.send(new SendMessageCommand({ + QueueUrl: requireEnv("AI_JOBS_QUEUE_URL"), + MessageBody: JSON.stringify({ run_id }), + MessageGroupId: undefined, + })); + } catch (sqsErr) { + // Without a message the run would sit in 'awaiting_ai' until + // reconcile-runs fails it, so fail it now (src/lib/queue-failure.ts). + log.error("AI jobs SQS send failed; marking run failed", { run_id, err: sqsErr }); + await failRunAfterQueueError(run_id, "awaiting_ai", AI_QUEUE_FAILED_MESSAGE); + return queueErrorResponse(AI_QUEUE_FAILED_MESSAGE); + } log.info("enqueued AI job", { run_id }); return NextResponse.json({ ok: true, queued: true }); } diff --git a/src/lib/queue-failure.ts b/src/lib/queue-failure.ts new file mode 100644 index 0000000..7d21828 --- /dev/null +++ b/src/lib/queue-failure.ts @@ -0,0 +1,48 @@ +/** + * What a route does when SendMessage to the executor queue or the AI jobs + * queue fails after it has moved a run forward (to 'queued' or 'awaiting_ai'). + * + * Without a message nothing would ever pick the run up: the reconcile-runs + * cron only marks it failed 20 minutes later. So the route fails the run now, + * with a message the run page shows, and answers 503. + * + * The UPDATE is conditional on the status the route set. SendMessage can fail + * after SQS accepted the message (a timeout), and if a worker has already + * claimed the run (for example 'queued' -> 'running') it is left alone. + */ +import { NextResponse } from "next/server"; +import { queryOne } from "@/lib/db"; +import { logger } from "@/lib/logger"; + +const log = logger.child({ module: "queue-failure" }); + +export const EXECUTOR_QUEUE_FAILED_MESSAGE = + "Could not queue this run for execution (the job queue was unavailable). The run was marked failed; upload the file again to retry."; + +export const AI_QUEUE_FAILED_MESSAGE = + "Could not queue this run for AI suggestions (the job queue was unavailable). The run was marked failed; upload the file again to retry."; + +/** Marks the run failed if it is still in `fromStatus`. Returns whether it did. Never throws. */ +export async function failRunAfterQueueError( + runId: string, + fromStatus: "queued" | "awaiting_ai", + message: string +): Promise { + try { + const row = await queryOne<{ id: string }>( + `UPDATE pipeline_runs SET status = 'failed', error_message = $3, updated_at = now() + WHERE id = $1 AND status = $2 + RETURNING id`, + [runId, fromStatus, message] + ); + return !!row; + } catch (err) { + // The reconcile-runs cron still fails the run after 20 minutes. + log.error("could not mark run failed after SQS send error", { run_id: runId, err }); + return false; + } +} + +export function queueErrorResponse(message: string): NextResponse { + return NextResponse.json({ error: message, run_status: "failed" }, { status: 503 }); +} diff --git a/tests/integration/approve-rules-claim.test.ts b/tests/integration/approve-rules-claim.test.ts index 73edf18..6869830 100644 --- a/tests/integration/approve-rules-claim.test.ts +++ b/tests/integration/approve-rules-claim.test.ts @@ -69,6 +69,25 @@ describe("approve-rules claim", () => { expect(await receive(env("SQS_QUEUE_URL"), (m) => JSON.parse(m.Body!).run_id === runId, 1, 3)).toHaveLength(1); }); + it("marks the run failed with a message when SendMessage really fails", async () => { + const user = newUser(); + authState.userId = user; + const { runId, ruleIds } = await seedAwaitingApproval(user); + const body = { run_id: runId, rule_decisions: ruleIds.map((rule_id) => ({ rule_id, action: "approved", modifications: null })) }; + const queueUrl = process.env.SQS_QUEUE_URL; + process.env.SQS_QUEUE_URL = queueUrl!.replace(/[^/]+$/, "cleanstack-test-no-such-queue"); + try { + const res = await approveRules(request("/api/approve-rules", { method: "POST", body })); + expect(res.status).toBe(503); + expect(await res.json()).toMatchObject({ run_status: "failed" }); + } finally { + process.env.SQS_QUEUE_URL = queueUrl; + } + const run = await db.query("SELECT status, error_message FROM pipeline_runs WHERE id = $1", [runId]); + expect(run.rows[0].status).toBe("failed"); + expect(run.rows[0].error_message).toMatch(/Could not queue this run for execution/); + }); + it("another team cannot approve the run", async () => { const owner = newUser(); const { runId, ruleIds } = await seedAwaitingApproval(owner);