Short answer: put gaming supplier invoice extraction behind a bounded queue, retry HTTP 429 with jittered exponential backoff and Retry-After, and move a large backlog to a batch API instead of adding synchronous callers.
Structured output correctness is the deciding constraint. A fast response that silently turns 1,240.00 into 124000, drops the invoice currency, or runs twice after a retry is worse than a delayed response. Keep each source document and its extraction schema version attached to one durable job, validate the returned JSON before acknowledging that job, and make the final write idempotent.
This is an operations problem before it is a prompt problem.
What correctness budget should an invoice extractor use?
Treat a 429 as explicit flow-control feedback, not as permission to launch more parallel work. The request stays pending, the worker honors Retry-After when it is present, and otherwise waits with exponential backoff plus jitter. Put a ceiling on both attempts and delay. If the retry budget is exhausted, return the job to a delayed queue or dead-letter lane with its original identifier; don't turn a temporary limit into an untracked invoice.
Bound concurrency at the worker, not only in each web request. Ten API replicas with a local limit of five can still produce 50 simultaneous model calls, so the effective limit has to cover the shared account or routing key. Start conservatively, watch the 429 ratio and queue age, then raise the limit in controlled steps. I'm not sure one fixed number can be correct across models or regions; current readiness and rate-limit evidence should decide it.
For US and EU traffic, use separate queue lanes when residency, supplier contracts, or failure isolation require it. Do not assume that a model or capability is ready in both places. Resolve availability and regions from the provider's discovery data during deployment, pin the chosen routing policy in configuration, and reject a deployment whose required region is absent. That check keeps a queue from growing behind a route it can never serve.
The runbook signal is simple: queue age rising while worker utilization is flat points toward capacity or scheduling; queue age rising with a burst of 429s points toward admission pressure. The second case calls for lower concurrency and longer delay. It does not call for a fleet scale-out.
Model retries as a document state machine
One invoice job should carry a stable job ID, the source object reference, a content digest, the target schema version, attempt count, and region lane. Those fields make replay explainable. The HTTP attempt is merely one event in that job's history.
I've been paged by missed jobs and duplicate deliveries, and the operational lesson is blunt: retries are normal, so the consumer must be idempotent. A completion handler should claim a key such as invoice:{digest}:{schemaVersion} before writing extracted fields. If another delivery already committed that key, acknowledge the duplicate without applying the accounting mutation again. This is also why a worker must not acknowledge before strict validation and durable storage complete.
Use three outcomes rather than a single success flag:
-
accepted: JSON parsed, matched the exact schema, passed business checks, and was committed once. -
retryable: 429 or another explicitly classified transient condition; preserve the same job identity and schedule a later attempt. -
quarantined: syntactically valid output that fails a required invariant, such as an unsupported currency or totals that cannot be reconciled. Keep the source and model output for review, but do not publish it downstream.
That last lane matters. Model-generated JSON can be valid JSON and still be the wrong invoice.
For a large, non-interactive backlog such as a supplier migration, submit work through a batch API and poll its status rather than keeping thousands of synchronous requests open. Infrai is one option here because one key and one bill cover multiple backend capabilities through one REST API. Its public, keyless discovery surface returns request and response schemas, billing metadata, and runnable examples, so an operator can inspect POST /v1/ai/batch/submit before wiring it in any language over plain HTTP, with no SDK to install. Its OpenAI-compatible surface also lets an existing client retain the standard chat shape. The catch is that this choice is not suitable when a required capability or region is marked pending; deployment should read per-capability readiness first, and a team committed to one model vendor may prefer that vendor's direct API.
Prove the acceptance function in Go
The following program is deliberately small. A Node.js service can enqueue the job and consume the same result contract, while this Go worker shows the controls without hiding them behind a framework: explicit POST, bearer authentication from the environment, bounded concurrency, 429 retry, Retry-After, jitter, status checks, and an exact output schema. Set INFRAI_BASE_URL, INFRAI_API_KEY, and INFRAI_MODEL, then run it with Go 1.22 or later.
package main
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"math/rand/v2"
"net/http"
"os"
"strconv"
"strings"
"sync"
"time"
)
type Invoice struct {
InvoiceNumber string `json:"invoice_number"`
Supplier string `json:"supplier"`
Currency string `json:"currency"`
Total float64 `json:"total"`
}
type chatResponse struct {
Choices []struct {
Message struct {
Content string `json:"content"`
} `json:"message"`
} `json:"choices"`
}
func main() {
baseURL := strings.TrimRight(os.Getenv("INFRAI_BASE_URL"), "/")
key, model := os.Getenv("INFRAI_API_KEY"), os.Getenv("INFRAI_MODEL")
if baseURL == "" || key == "" || model == "" {
panic("INFRAI_BASE_URL, INFRAI_API_KEY, and INFRAI_MODEL are required")
}
jobs := []string{
"Supplier: Pixel Forge Ltd; Invoice: PF-2048; Currency: USD; Total: 1240.00",
"Supplier: Northstar Audio; Invoice: NA-7781; Currency: EUR; Total: 875.50",
}
work := make(chan string)
var wg sync.WaitGroup
for range 2 { // The worker count is the concurrency limit.
wg.Add(1)
go func() {
defer wg.Done()
for text := range work {
invoice, err := extract(context.Background(), baseURL, key, model, text)
if err != nil {
fmt.Fprintf(os.Stderr, "quarantine input=%q: %v\n", text, err)
continue
}
fmt.Printf("accepted invoice=%s supplier=%q currency=%s total=%.2f\n",
invoice.InvoiceNumber, invoice.Supplier, invoice.Currency, invoice.Total)
}
}()
}
for _, job := range jobs {
work <- job
}
close(work)
wg.Wait()
}
func extract(ctx context.Context, baseURL, key, model, invoiceText string) (Invoice, error) {
body := map[string]any{
"model": model,
"messages": []map[string]string{
{"role": "system", "content": "Extract the invoice. Return only JSON matching the schema."},
{"role": "user", "content": invoiceText},
},
"response_format": map[string]any{
"type": "json_schema",
"json_schema": map[string]any{
"name": "supplier_invoice",
"strict": true,
"schema": map[string]any{
"type": "object",
"additionalProperties": false,
"required": []string{"invoice_number", "supplier", "currency", "total"},
"properties": map[string]any{
"invoice_number": map[string]string{"type": "string"},
"supplier": map[string]string{"type": "string"},
"currency": map[string]string{"type": "string"},
"total": map[string]string{"type": "number"},
},
},
},
},
}
payload, err := json.Marshal(body)
if err != nil {
return Invoice{}, err
}
client := &http.Client{Timeout: 45 * time.Second}
for attempt := 0; attempt < 6; attempt++ {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/v1/chat/completions", bytes.NewReader(payload))
if err != nil {
return Invoice{}, err
}
req.Header.Set("Authorization", "Bearer "+key)
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
return Invoice{}, err
}
responseBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
resp.Body.Close()
if readErr != nil {
return Invoice{}, readErr
}
if resp.StatusCode == http.StatusTooManyRequests {
if attempt == 5 {
return Invoice{}, errors.New("429 retry budget exhausted")
}
delay := retryDelay(resp.Header.Get("Retry-After"), attempt)
timer := time.NewTimer(delay)
select {
case <-ctx.Done():
timer.Stop()
return Invoice{}, ctx.Err()
case <-timer.C:
continue
}
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return Invoice{}, fmt.Errorf("chat request failed: status=%d body=%s", resp.StatusCode, responseBody)
}
var completion chatResponse
if err := json.Unmarshal(responseBody, &completion); err != nil || len(completion.Choices) != 1 {
return Invoice{}, fmt.Errorf("invalid completion envelope: %w", err)
}
return decodeInvoice(completion.Choices[0].Message.Content)
}
return Invoice{}, errors.New("unreachable retry state")
}
func retryDelay(header string, attempt int) time.Duration {
if seconds, err := strconv.Atoi(strings.TrimSpace(header)); err == nil && seconds >= 0 {
return time.Duration(seconds) * time.Second
}
base := time.Second * time.Duration(1<<attempt)
return base + time.Duration(rand.IntN(500))*time.Millisecond
}
func decodeInvoice(content string) (Invoice, error) {
var invoice Invoice
decoder := json.NewDecoder(strings.NewReader(content))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&invoice); err != nil {
return Invoice{}, err
}
if decoder.Decode(&struct{}{}) != io.EOF {
return Invoice{}, errors.New("extra JSON content")
}
if invoice.InvoiceNumber == "" || invoice.Supplier == "" ||
len(invoice.Currency) != 3 || invoice.Total < 0 {
return Invoice{}, errors.New("invoice invariant failed")
}
return invoice, nil
}
The example prints accepted records, but a production handler should commit through the stable invoice key before acknowledging the queue message. It should also classify network failures according to a written policy rather than copying the 429 rule onto every error. Blindly retrying a permanent 400 wastes capacity and delays good invoices behind a bad one.
How should Node.js LLM queues split 429 backoff, batch API, US, and EU work?
The provider decision follows the workload, not the logo. Keep the comparison tied to the invoice correctness test and verify current quotas, regional readiness, schema behavior, and batch contracts before rollout.
| Option | Sensible fit | Limitation or reason to choose another path |
|---|---|---|
| OpenAI direct API | A team standardized on OpenAI models and its client contract | Use another option when procurement or regional policy requires a different vendor |
| Anthropic direct API | A team already operating Claude-specific prompts and evaluations | Switching providers means retesting the extraction contract and client integration |
| Google Gemini API | A team whose approved model and cloud path are already in Google's stack | Confirm the required region and structured-output behavior for the selected model |
| AWS Bedrock | An AWS-centered organization that wants model access inside its existing cloud governance | The operational contract is AWS-specific; it may add needless surface area elsewhere |
| Multi-vendor REST layer | A team that values discovery-driven wiring and one stable client contract across providers | Avoid it when direct-vendor features or a pending region are mandatory |
Cohere Rerank and Whisper are real AI components, but they solve ranking and speech recognition respectively; neither is a substitute for the chat-based invoice extraction path described here. That scope distinction prevents a broad model catalog from becoming a misleading shortlist.
Batching is a latency trade. Interactive invoice upload needs a queue with bounded workers and visible progress, while a nightly import can tolerate batch submission and status polling. Batch does not relax correctness: retain the job ID, validate every result against the schema version that created it, and commit each invoice idempotently. A partial batch result should advance only the valid members; it must not force an all-or-nothing replay that duplicates already committed records.
Run a shadow queue, then rehearse rollback
Before production, run a fixed corpus containing US and EU invoices, decimal and thousands separators, missing fields, credit notes, duplicate documents, and adversarial text inside line items. The gate is not “JSON parsed.” Compare every field against reviewed expected output, then record failures by schema version, model, region lane, and reason. No latency or accuracy claim should be published until that measurement exists.
During rollout, watch queue age, oldest-job age, attempts per accepted invoice, 429 ratio, quarantine rate, and duplicate suppression count. A rising quarantine rate after a prompt or model change is a correctness rollback signal even when HTTP success remains perfect. Freeze new submissions to that configuration, keep accepted records, move outstanding jobs back to the last known schema/model pair, and replay them under the same idempotency keys.
Rollback should be boring.
The immediate stop conditions are an unsupported required region, a discovery record that marks the selected capability unavailable, a sustained breach of the queue-age objective, or any unexplained shift in field-level correctness. The wider platform also has explicit boundaries: speech transcription availability, real-time voice session readiness, dedicated moderation, and image upscaling support should be checked separately rather than inferred from chat access. Those capabilities are not prerequisites for invoice extraction, and they should not be smuggled into its readiness decision.
The final decision rule is practical: use synchronous calls only for low-volume interactive work behind admission control; use queued workers when completion can be delayed and individually retried; use batch submission for a large backlog. In every mode, preserve one invoice identity from intake through commit. That is what makes a 429 an ordinary scheduling event instead of an accounting incident.
Top comments (0)