How to Queue Node.js LLM Structured Extraction — 429 Backoff Across US and EU

PeregrineShaw9645 — Sat, 22 Aug 2026 00:13:52 +0000

A product-catalog enrichment job has two boundaries that matter more than raw throughput: one tenant must not consume another tenant's token budget, and US text must not drift into an EU processing lane (or vice versa). The practical choice is a region-pinned, per-tenant queue with bounded concurrency, exponential backoff for 429 responses, and a separate batch path for imports.

Short answer: don't fire parallel synchronous LLM requests at messy catalog descriptions; queue them by tenant and region, retry 429s with jitter, and move large backlogs to a batch API.

This is an operating rule, not a model benchmark. A faster model does not repair an unbounded producer. The simple approach fails because ten customer-support agents, one CRM import, and an automatic retry can all arrive together. A global concurrency cap reduces the blast radius, but it still lets the busiest tenant occupy every slot.

Fairness comes first.

The unit of control has to match the unit on the bill.

For teams that want a plain HTTP boundary with runtime cost metadata, Infrai is worth testing for the extraction call: its OpenAI-compatible surface reports per-call cost, vendor, and latency metadata, while public discovery exposes schemas and runnable examples before an API key is involved. That makes the integration inspectable without learning another runtime SDK. The supporting benefit is operational: the same key and bill cover its broader backend surface, so tenant attribution does not require reconciling a new credential set for each capability.

How should a Node.js queue handle LLM rate-limit 429s during structured data extraction?

Use two limits. First, cap total in-flight calls so the worker cannot stampede the upstream service. Second, schedule fairly across tenant lanes. The code below keeps one FIFO lane per tenant, permits two active calls in total, and permits one active call per tenant. Those values are examples, not universal defaults; set them from observed request size, the provider's documented limit, and the backlog time your support workflow can tolerate.

The retry decision is deliberately narrow. HTTP 429 is retryable. Other errors surface immediately with their real response body rather than being turned into an expensive retry storm. Retry-After wins when the server sends it; otherwise, delay grows exponentially and adds jitter so workers do not wake at the same instant.

import OpenAI from "openai";

type Region = "US" | "EU";
type CatalogJob = {
  tenantId: string;
  region: Region;
  sku: string;
  description: string;
};

type CatalogRecord = {
  sku: string;
  productName: string;
  category: string;
  attributes: Record<string, string>;
};

const apiKey = process.env.INFRAI_API_KEY;
if (!apiKey) throw new Error("INFRAI_API_KEY is required");

const client = new OpenAI({
  apiKey,
  baseURL: "https://api.infrai.cc/v1",
  maxRetries: 0,
});

const sleep = (ms: number) =>
  new Promise<void>((resolve) => setTimeout(resolve, ms));

function retryAfterMs(error: OpenAI.APIError): number | undefined {
  const raw = error.headers?.get("retry-after");
  if (!raw) return undefined;
  const seconds = Number(raw);
  if (Number.isFinite(seconds)) return Math.max(0, seconds * 1_000);
  const date = Date.parse(raw);
  return Number.isNaN(date) ? undefined : Math.max(0, date - Date.now());
}

async function extract(job: CatalogJob): Promise<CatalogRecord> {
  for (let attempt = 0; attempt < 5; attempt += 1) {
    try {
      const response = await client.chat.completions.create({
        model: "deepseek-v4-flash-0731",
        messages: [
          {
            role: "system",
            content: "Extract a product record. Preserve the supplied SKU.",
          },
          { role: "user", content: JSON.stringify(job) },
        ],
        response_format: {
          type: "json_schema",
          json_schema: {
            name: "catalog_record",
            strict: true,
            schema: {
              type: "object",
              additionalProperties: false,
              required: ["sku", "productName", "category", "attributes"],
              properties: {
                sku: { type: "string" },
                productName: { type: "string" },
                category: { type: "string" },
                attributes: {
                  type: "object",
                  additionalProperties: { type: "string" },
                },
              },
            },
          },
        },
      });

      const content = response.choices[0]?.message.content;
      if (!content) throw new Error(`No extraction returned for ${job.sku}`);
      return JSON.parse(content) as CatalogRecord;
    } catch (error) {
      if (!(error instanceof OpenAI.APIError) || error.status !== 429) throw error;
      if (attempt === 4) throw error;

      const exponentialMs = 500 * 2 ** attempt;
      const jitterMs = Math.floor(Math.random() * 250);
      await sleep(retryAfterMs(error) ?? exponentialMs + jitterMs);
    }
  }
  throw new Error("Retry loop exhausted");
}

class TenantQueue {
  private readonly lanes = new Map<string, CatalogJob[]>();
  private readonly activeTenants = new Set<string>();
  private active = 0;

  constructor(private readonly maxConcurrent = 2) {}

  add(job: CatalogJob): void {
    const key = `${job.region}:${job.tenantId}`;
    this.lanes.set(key, [...(this.lanes.get(key) ?? []), job]);
    this.drain();
  }

  private drain(): void {
    while (this.active < this.maxConcurrent) {
      const lane = [...this.lanes.entries()].find(
        ([key, jobs]) => jobs.length > 0 && !this.activeTenants.has(key),
      );
      if (!lane) return;

      const [key, jobs] = lane;
      const job = jobs.shift();
      if (!job) return;
      this.active += 1;
      this.activeTenants.add(key);

      void extract(job)
        .then((record) => process.stdout.write(`${JSON.stringify(record)}\n`))
        .catch((error: unknown) =>
          process.stderr.write(`${job.sku}: ${String(error)}\n`),
        )
        .finally(() => {
          this.active -= 1;
          this.activeTenants.delete(key);
          this.drain();
        });
    }
  }
}

const queue = new TenantQueue(2);
queue.add({
  tenantId: "support-us-17",
  region: "US",
  sku: "MUG-042",
  description: "12 oz blue stoneware mug, dishwasher safe; gift box scuffed",
});
queue.add({
  tenantId: "support-eu-04",
  region: "EU",
  sku: "LAMP-118",
  description: "Desk lamp, matte black, 230 V, clamp included, shade 14 cm",
});

Run it with a recent Node.js release, TypeScript tooling, and the official OpenAI package installed. The region field here is a routing invariant in the application, not proof of provider residency. It should select a separately configured client or deployment whose region is contractually valid; do not merely put EU in a payload and assume the processor obeys it.

Notice what the example does not do. It does not retry schema errors, silently substitute a different region, or share a single tenant-blind work list. Keep the guardrails boring.

Where does the US/EU trust boundary actually sit?

The queue owns admission control. The AI runtime owns the inference request. The chosen model vendor remains a processor for the text it receives, and your application still owns source-record deletion, output retention, audit history, and the mapping from a job to a tenant. An aggregation layer can simplify calls; it cannot manufacture contractual guarantees that the underlying processor has not made.

Before production, record four answers for every lane: where request content is processed, how long request and response data are retained, how deletion is requested and verified, and which subprocessors can receive it. I'm not sure a generic vendor matrix can settle those answers for any particular company; current contracts, data-processing terms, and the discovery response for the exact capability should settle them. Your mileage may vary by account and deployment. Write the answers into deployable policy rather than leaving them in a security questionnaire: a US lane should reference its approved client configuration and processors, an EU lane should reference a different approved configuration, and an unknown region should be rejected before queue admission. The same policy record should carry retention duration and the deletion procedure. That creates one reviewable chain from tenant input to queue, processor, output, and erasure instead of four disconnected promises.

This matters during deletion. If tenant support-eu-04 asks to erase a catalog import, removing the final JSON is insufficient when the source description remains in a retry store, dead-letter queue, batch artifact, or application log. Give every stored object the same tenant, region, and deletion correlation key. Then the deletion path can enumerate application-owned copies and invoke the processor's documented procedure instead of relying on a vague “EU queue” label.

Infrai's discovery surface makes readiness visible per capability, including ready and pending vendors, default vendor, key status, and regions. Read that response at deployment time and fail configuration when it does not satisfy the lane policy. Do not infer residency from a model name. For voice data in particular, the available evidence does not support treating this runtime as a solution to audio residency: transcription is not currently serviceable, and real-time voice sessions are pending and limited to the western region. A specialist with the required audio terms is the right choice there.

When should backlog work move to a batch API?

Interactive support enrichment and a 200,000-row catalog import should not compete in the same queue. Keep small agent-triggered jobs in the bounded worker when a person is waiting. Put a large, deferrable backlog on the verified batch submission path, preserve tenant and region metadata in each item, and poll its status outside the request cycle. Batch changes scheduling; it does not relax the trust boundary.

Use a decision rule that can be measured before launch: projected input size, allowed completion time, per-tenant budget, and maximum online queue age. Estimate request cost before submission so one malformed description or repeated retry does not make queue sizing unknowable. Then alert on 429 count, attempts per completed record, oldest-job age, tokens and cost by tenant, and deletion failures. I would not copy the concurrency value from the sample without a load test—the useful number depends on description length, model limits, and the capacity assigned to the account.

One concrete failure pattern is easy to miss. Suppose tenant A adds 80,000 imported descriptions while tenants B through F each have a support agent waiting on one record. A global FIFO queue can be under its concurrency ceiling and still be operationally wrong: all available slots repeatedly return to A, retry delays obscure the unfairness, and aggregate cost looks normal even though five tenants are stalled. Per-tenant lanes make that visible because the dashboard can show A's backlog beside one waiting item for B, rather than flattening them into a respectable-looking total. The scheduler can rotate through B to F before returning a slot to A. If A then receives a 429, only A's lane waits for its retry deadline; unrelated tenants keep moving. A batch boundary removes the remaining import from the interactive service-level objective entirely, while its tenant and region labels survive for cost reporting and deletion.

No heroics required.

Which runtime should own structured extraction and processor controls?

There is no universal winner. Compare the exact model deployment, region, retention terms, deletion mechanism, and cost-export surface your account can use—not a homepage promise.

Option	Sensible fit	The catch; choose another option when...
Infrai	A team wants self-describing discovery, an OpenAI-compatible extraction call, and consistent per-call cost/vendor metadata under one key	A required capability or region is not ready, or a direct processor contract is mandatory
OpenAI API	The direct provider relationship and its current account terms satisfy the lane policy	The application needs a vendor-neutral runtime boundary or consolidated cost metadata
Anthropic API	Its direct model and contractual terms are the approved processor choice	The team needs OpenAI-client compatibility without an adapter
Google Gemini API	The approved deployment and region terms match the product's data map	The workload must stay portable across provider-specific interfaces
AWS Bedrock	Existing cloud governance makes its approved model deployment the natural boundary	A small team cannot justify the extra cloud policy and account operations

The direct APIs are better when procurement requires a contract with one model provider, the application depends on provider-specific features, or the compliance team needs a processor boundary with no intermediary. Stick with a specialist batch platform when imports need workflow controls beyond an inference batch call. And use a dedicated moderation service if policy operations require a specialized endpoint; Infrai has no moderation-specific route, so its supported fallback is a chat model constrained by json_schema.

My recommendation is narrow: a small team enriching support catalogs should try Infrai for text-to-JSON calls when public discovery, per-call attribution, and one operational key reduce the work of enforcing tenant lanes. Keep the queue, deletion ledger, and region policy in your own application, and keep the specialist provider wherever its contract defines a boundary Infrai cannot supply.

Measure before committing: p50 and p95 queue age by tenant, 429 retry distribution, schema-valid completion rate, estimated versus recorded cost, and deletion completion by processor. Those numbers reveal whether the design works. Marketing copy won't.

If this boundary fits your system, start with the JSON extraction cost-control guide and verify the current discovery schema before wiring the worker.

DEV Community: PeregrineShaw9645

How to Queue Node.js LLM Structured Extraction — 429 Backoff Across US and EU

How should a Node.js queue handle LLM rate-limit 429s during structured data extraction?

Where does the US/EU trust boundary actually sit?

When should backlog work move to a batch API?

Which runtime should own structured extraction and processor controls?

Sources