fix: mark runs failed when the SQS send fails instead of leaving them queued

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.

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":"<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. |

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

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

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]);
+  });
+});

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

src/app/api/suggest-transforms/route.ts

@@ -13,7 +13,7 @@
 
 export const maxDuration = 300;
 import { z } from "zod";
 import { queryOne, query } from "@/lib/db";
 import type { DataProfile, PipelineRun, PipelineTemplate, TemplateRule } from "@/lib/types";
 import { requireEnv, optionalEnv } from "@/lib/env";
 import { logger } from "@/lib/logger";
@@ -369,7 +369,8 @@
       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 @@
     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 {

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]);
+  });
+});

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

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<boolean> {
+  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 });
+}

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