Message too large and JSON that won't parse: fixing malformed queue payloads

The 256 KB ceiling, the object-only payload rule, and the schema check your producer owes the worker — with the exact errors Infrai's queue returns and the claim-check fix.

“Malformed payload” covers three unrelated bugs, and they need three different fixes: a message that exceeds the broker’s size ceiling, a request body that isn’t valid JSON at all, and a payload that parses cleanly but doesn’t contain the fields your worker expects. Infrai’s queue enforces the first two at publish time and leaves the third entirely to you, which is the right split — the broker can’t know your schema.

Size first, because it’s the one that shows up in production rather than in development.

Every queue has a 256 KB ceiling, and it’s on the message

The limit is an attribute of the queue itself, visible the moment you create one:

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X POST "https://api.infrai.cc/v1/queue/create" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"name":"doc-ingest","type":"standard","dlq":"doc-ingest-dlq"}'

curl -sS "https://api.infrai.cc/v1/queue/get/doc-ingest" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

{
  "ok": true,
  "data": {
    "name": "doc-ingest",
    "type": "standard",
    "message_retention_days": 14,
    "max_message_size_kb": 256,
    "visibility_timeout_default": 300,
    "delivery_delay_seconds": 0,
    "max_receive_count": 3,
    "dlq_name": "doc-ingest-dlq"
  }
}

256 KB is 262,144 bytes of serialized payload, and it’s the same ceiling SQS applies to a single message — a number worth internalising because it’s smaller than one page of scraped HTML with its markup intact. Testing the boundary on 2026-07-26, a 255 KB payload published normally and a 258 KB one came back as HTTP 400.

Here’s the part that will cost you an afternoon if nobody warns you. The oversize rejection doesn’t say anything about size:

{
  "ok": false,
  "error": {
    "code": "INVALID_ARGUMENT",
    "http_status": 400,
    "message": "queue.publish failed: queue '4ee0c441fa36c267 doc-ingest' already exists",
    "retryable": false,
    "trace_id": "trc_cef2153df8284f2d87c9374f"
  }
}

The queue exists, obviously; the message text is wrong and the QUEUE_MESSAGE_TOO_LARGE code it should carry never appears. Treat any INVALID_ARGUMENT on publish whose text mentions the queue already existing as a size failure until that’s fixed, and measure client-side so you never see it.

Put the bytes somewhere else and publish the pointer

The pattern is older than any of these products: store the blob, queue the reference. A job message should be an instruction, not a document.

import process from "node:process";
import { Buffer } from "node:buffer";

const BASE = "https://api.infrai.cc";
const QUEUE = "doc-ingest";
const MAX_BYTES = 256 * 1024;
const SAFE_BYTES = 240 * 1024;
const apiKey = process.env.INFRAI_API_KEY;
if (!apiKey) throw new Error("INFRAI_API_KEY is required");

const REQUIRED = { document_id: "string", source_url: "string", tenant: "string", pages: "number" };

function validate(payload) {
  const problems = [];
  for (const [field, type] of Object.entries(REQUIRED)) {
    if (!(field in payload)) problems.push(`missing ${field}`);
    else if (typeof payload[field] !== type) problems.push(`${field} must be ${type}, got ${typeof payload[field]}`);
  }
  if (payload.pages !== undefined && payload.pages < 1) problems.push("pages must be >= 1");
  return problems;
}

export async function enqueueDocument(payload) {
  const problems = validate(payload);
  if (problems.length) throw new Error(`refusing to publish: ${problems.join("; ")}`);

  const bytes = Buffer.byteLength(JSON.stringify(payload), "utf8");
  if (bytes > SAFE_BYTES) {
    throw new Error(`payload is ${bytes} bytes, over the ${SAFE_BYTES} byte publish budget (queue ceiling ${MAX_BYTES}) — upload the body and publish its key instead`);
  }

  const res = await fetch(`${BASE}/v1/queue/publish`, {
    method: "POST",
    headers: { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" },
    body: JSON.stringify({ queue: QUEUE, body: payload }),
  });
  const json = await res.json();
  if (!res.ok || json.ok === false) {
    const err = json.error ?? {};
    throw new Error(`publish rejected (${err.code ?? res.status}): ${err.message ?? "no detail"}`);
  }
  for (const warning of json.metadata?.warnings ?? []) console.warn(`api warning: ${warning}`);
  return json.data.message_id;
}

const id = await enqueueDocument({
  document_id: "doc_5591",
  source_url: "https://example.com/reports/q2.html",
  tenant: "acme",
  pages: 34,
});
console.log(`queued ${id}`);

Two things that block a whole class of incidents. The 240 KB budget leaves headroom, because the field you add next month is the one that crosses the line. And the loop over metadata.warnings surfaces messages you’d otherwise never read — publish answers 'body' is not a field of this endpoint; interpreted it as 'payload', which tells you the documented alias is being mapped rather than silently dropped.

For a real schema, reach for a library. Ajv compiles a JSON Schema once and validates in microseconds; Zod is nicer if your producer is TypeScript. The hand-rolled version above is here so the example runs with no install.

What the API rejects, and who should have caught it

The bottom two rows are the whole argument for validating before you publish. A queue that accepts anything JSON-shaped will happily store a job your worker can never run.

You can watch the first row happen with a deliberately truncated body:

curl -sS -X POST "https://api.infrai.cc/v1/queue/publish" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  --data-raw '{"queue":"doc-ingest","body":{"document_id":"doc_1"'

{
  "ok": false,
  "error": {
    "code": "INVALID_ARGUMENT",
    "http_status": 400,
    "message": "request body must be valid JSON",
    "code_detail": "invalid_json",
    "retryable": false
  }
}

retryable: false is the field to branch on in a producer wrapper — retrying a syntax error just burns rate limit.

The consumer still has to defend itself

Old messages predate your latest schema, and a bad deploy can publish garbage for ten minutes before anyone notices. A worker that throws on a malformed message and doesn’t ack turns one bad payload into three redeliveries and a dead-letter entry, which is fine — but only if the failure is genuinely transient. A message that will never be valid should be acked and recorded, not retried.

import process from "node:process";

const BASE = "https://api.infrai.cc";
const QUEUE = "doc-ingest";
const apiKey = process.env.INFRAI_API_KEY;
if (!apiKey) throw new Error("INFRAI_API_KEY is required");
const headers = { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" };

async function call(path, payload) {
  const res = await fetch(`${BASE}${path}`, { method: "POST", headers, body: JSON.stringify(payload) });
  const json = await res.json();
  if (!res.ok || json.ok === false) throw new Error(`${path}: ${json?.error?.message ?? res.status}`);
  return json.data;
}

function shapeErrors(payload) {
  if (payload === null || typeof payload !== "object" || Array.isArray(payload)) return ["payload is not an object"];
  const missing = ["document_id", "source_url", "tenant"].filter((f) => typeof payload[f] !== "string");
  return missing.map((f) => `missing or non-string ${f}`);
}

export async function work() {
  const { items } = await call("/v1/queue/consume", { queue: QUEUE, max_messages: 10 });
  for (const message of items) {
    const errors = shapeErrors(message.payload);
    if (errors.length) {
      console.error(`poison ${message.message_id} (delivery ${message.delivery_count}): ${errors.join(", ")}`);
      await recordQuarantine(message, errors);
      await call("/v1/queue/ack", { queue: QUEUE, receipt_handle: message.message_id });
      continue;
    }
    try {
      await ingest(message.payload);
      await call("/v1/queue/ack", { queue: QUEUE, receipt_handle: message.message_id });
    } catch (err) {
      console.error(`transient failure on ${message.message_id}: ${err.message} — leaving it unacked`);
    }
  }
  return items.length;
}

async function recordQuarantine(message, errors) {
  console.log(JSON.stringify({ quarantined: message.message_id, errors, payload: message.payload }));
}

async function ingest(payload) {
  console.log(`ingesting ${payload.document_id} for ${payload.tenant} from ${payload.source_url}`);
}

await work();

Unacked messages come back three times — max_receive_count is 3 — and then land in doc-ingest-dlq. Check where you stand:

curl -sS "https://api.infrai.cc/v1/queue/stats/doc-ingest" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

A non-zero dlq_count after a deploy is usually a schema change that shipped to the producer before the consumer.

Cost, and where this queue isn’t the answer

Publishing is the only metered call, at $0.00002 per message, verified 2026-07-26; consume, ack, stats, create and dead-letter reads are free and rate-limited. Rejected publishes aren’t billed, and rates here move downward over time, so check rather than budget from this page:

curl -sS "https://api.infrai.cc/v1/discovery" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | jq '[.capabilities[] | select(.id | startswith("queue.")) | {id, billable: .billing.is_billable, usd: .billing.price_usd}]'

New accounts carry $2 of credit, which is a lot of test publishes.

If your payloads genuinely exceed 256 KB and you can’t split them, SQS with the Extended Client Library automates the store-in-S3-publish-the-pointer dance, and Kafka’s default 1 MB record limit is configurable upward — you’d be better off there for streaming raw events. BullMQ keeps job data in Redis, so the practical limit is your memory budget rather than a fixed ceiling; the trade-off is running Redis. Infrai’s queue is the fit when jobs are small instructions, the workers speak HTTP, and the artifact storage, the notification email and the error report all sit on the same key and the same bill.

Two limitations to weigh: the size error message is currently wrong, and there’s no server-side schema registry, so nothing stops a stale producer from publishing a shape no worker understands.

References

• Infrai queue API reference
• QUEUE_MESSAGE_TOO_LARGE error reference
• Amazon SQS message quotas
• Ajv JSON Schema validator