Node.js LLM Structured Extraction Retries with Observable Idempotency for Supplier Invoices

#node #observability #llm

TL;DR: For Node.js LLM structured extraction, make retries safe with an idempotency key at the database commit, then track attempts separately from outcomes. Give each accepted supplier invoice one tenant-scoped operation ID and carry it across the webhook, worker, and JSON pipeline. This keeps a repeated call visible without letting it create duplicate records.

Pick this control	When it earns its keep	What it reveals	Main limit
Unique commit key	Every pipeline that writes extracted invoices	Duplicate commit attempts by tenant	It cannot explain why an attempt failed
Correlated structured logs	Low-to-moderate volume or early operations	The path of one invoice and its attempts	Cardinality and sensitive fields need discipline
Metrics by tenant and outcome	Shared workers with per-tenant budgets	Retry pressure, latency, and committed volume	Aggregates hide individual traces
Distributed traces	Calls cross queues, model gateways, and databases	Where time and errors accumulate	Sampling can omit rare paths

The decision rule is compact: enforce identity at the write boundary first, then instrument the same identity across the queue. Logs, metrics, and traces diagnose duplication. A unique database constraint prevents it. You need both.

Identity comes first.

How should a Node.js worker handle LLM structured extraction retries?

A queue normally promises delivery, not a single execution. A worker can finish the model call, write a row, and lose its acknowledgement before the broker sees it. The same message then runs again. A webhook sender can also repeat delivery after its own timeout. Neither event proves that the first attempt failed.

That creates three identities that teams often collapse into one: the business operation, the delivery, and the attempt. For supplier invoice extraction, the operation might be (tenant_id, source_system, source_invoice_id, extraction_schema_version). Delivery IDs belong to transport. Attempt numbers belong to execution. Only the operation identity should decide whether a second payable record may be committed. This distinction matters for cost visibility too. A tenant may have one committed invoice, two queue deliveries, and three model attempts. Reporting all three as "documents processed" hides retry cost. Reporting only the committed invoice hides operational waste. Keep both counters. Diagram in words: webhook to inbox row; inbox row to queue delivery; delivery to one or more extraction attempts; validated extraction to one guarded commit; commit to an acknowledgement. The operation ID crosses every arrow. If that ID disappears at any arrow, the dashboard can describe each local step while still failing to connect the expensive retry to the record it eventually produced.

Pick this when visibility is the immediate gap

Start with correlated structured logs when engineers cannot reconstruct a single invoice path. Log identifiers and state transitions, not raw invoice text or the full model response. OWASP describes sensitive-information disclosure as an LLM application risk; invoices can contain names, addresses, tax identifiers, and bank details, so payload logging expands the exposure surface.

A useful event is small: tenantId, operationId, deliveryId, attempt, stage, outcome, durationMs, and a stable error class. Keep model output out of it. Short logs win.

Retries are evidence.

Choose tenant-level metrics when the question is allocation: which tenant is generating attempts, validation failures, and committed results? Use bounded labels such as outcome and schema version. Do not place invoice IDs in metric labels; those belong in logs or traces. Measure model attempts and durable commits as different instruments, then derive attempts per commit over the same interval. A rising ratio is an alerting signal, not proof of duplicate rows.

Use distributed tracing once the path crosses enough process boundaries that timestamps are hard to join. Put the operation ID in span attributes and link redeliveries back to the operation. Traces explain time. The database still arbitrates ownership of the result.

One key. One commit.

Go deep on the guarded commit

The implementation below uses a generic store interface so the important contract stays visible. The insert must be atomic: create the result if its operation key is absent, or return the existing result. A preflight SELECT followed by an unconditional INSERT is not equivalent; two workers can both observe absence before either writes.

type Extraction = {
  supplierName: string;
  invoiceNumber: string;
  currency: string;
  totalMinor: number;
};

type Job = {
  tenantId: string;
  operationId: string;
  deliveryId: string;
  attempt: number;
  text: string;
};

type CommitResult =
  | { status: "inserted"; recordId: string }
  | { status: "existing"; recordId: string };

interface ResultStore {
  commitOnce(input: {
    tenantId: string;
    operationId: string;
    value: Extraction;
  }): Promise<CommitResult>;
}

interface Extractor {
  extract(text: string): Promise<unknown>;
}

interface Telemetry {
  event(name: string, fields: Record<string, string | number>): void;
  increment(name: string, fields: Record<string, string>): void;
}

Validation belongs before the guarded commit. Invalid structured output is an attempt outcome, not a partially useful invoice row. The worker records the attempt, asks the extractor for data, validates types and business invariants, and then calls the single atomic storage operation.

function validateExtraction(value: unknown): Extraction {
  if (typeof value !== "object" || value === null) {
    throw new Error("invalid_shape");
  }

  const row = value as Record<string, unknown>;
  if (
    typeof row.supplierName !== "string" ||
    typeof row.invoiceNumber !== "string" ||
    typeof row.currency !== "string" ||
    typeof row.totalMinor !== "number" ||
    !Number.isSafeInteger(row.totalMinor) ||
    row.totalMinor < 0
  ) {
    throw new Error("invalid_fields");
  }

  return row as Extraction;
}

async function processInvoice(
  job: Job,
  extractor: Extractor,
  store: ResultStore,
  telemetry: Telemetry
): Promise<void> {
  const startedAt = Date.now();
  telemetry.increment("extraction_attempts", { tenantId: job.tenantId });

  try {
    const value = validateExtraction(await extractor.extract(job.text));
    const commit = await store.commitOnce({
      tenantId: job.tenantId,
      operationId: job.operationId,
      value
    });

    telemetry.increment("extraction_commits", {
      tenantId: job.tenantId,
      outcome: commit.status
    });
    telemetry.event("invoice_extraction_finished", {
      tenantId: job.tenantId,
      operationId: job.operationId,
      deliveryId: job.deliveryId,
      attempt: job.attempt,
      outcome: commit.status,
      durationMs: Date.now() - startedAt
    });
  } catch (error) {
    telemetry.event("invoice_extraction_failed", {
      tenantId: job.tenantId,
      operationId: job.operationId,
      deliveryId: job.deliveryId,
      attempt: job.attempt,
      outcome: error instanceof Error ? error.message : "unknown_error",
      durationMs: Date.now() - startedAt
    });
    throw error;
  }
}

The store should enforce a unique key over tenant_id and operation_id. Include the tenant because supplier identifiers can collide across marketplace accounts. Include schema version in the operation ID when changing the extraction contract is meant to create a new result; otherwise a replay after a schema migration may silently return an older shape.

There is a deliberate trade-off here. Returning the existing record treats equivalent retries as success and lets the worker acknowledge the delivery. If the same operation ID arrives with materially different source content, silently returning the old value is dangerous. Store a source-content digest beside the result and classify a mismatched digest as an identity conflict for review. Do not make the digest from model output, because output can vary between attempts.

Alert on ratios, then inspect identities

A page for every retry will become noise. Alert on sustained changes in attempt outcomes and on invariant violations: commit conflicts with mismatched source digests, an absence of commits while accepted work continues, or attempts per commit moving outside the tenant's established operating band. The exact threshold must come from observed traffic and an error budget; no universal percentage is honest here.

Per-tenant cost visibility follows from the event model. Attribute model usage to an attempt, attribute a durable business result to a commit, and aggregate both by tenant. This answers two different questions: what computation did the tenant consume, and how many invoices became usable records? Never infer the first number from the second.

Test the boundary under concurrency. Send the same operation through several workers at once and assert one inserted result, with every other call receiving the same record ID as existing. Then test a redelivery after a simulated acknowledgement loss, a validation failure followed by a valid retry, and two tenants that share the same source invoice ID. These tests target the places where dashboards can look calm while duplicate writes still occur.

Limits to keep explicit

Idempotency cannot decide whether two different supplier documents represent the same real-world invoice. That requires a separate duplicate-detection policy, usually with review because invoice numbers and formatting can be inconsistent. Keep that fuzzy judgment away from the exact operation key.

Observability also cannot repair a non-atomic write path. It can expose existing outcomes, retry amplification, and identity conflicts. Prevention remains a storage invariant. Finally, retention and access controls matter: even identifier-only telemetry can reveal tenant activity patterns, so retain only what the operational questions require.

DEV Community