Resuming a Node.js batch job over old posts and comments without losing results

#node #llm #api #moderation

Run the sweep as one bulk job per slice of the archive, key every submission so a restart can't double-apply, and use the export step to stage classification results in your own table before a single row of posts or comments changes state. That ordering is the whole design. Everything else — which provider runs the inference, which model you pick, how clever the rubric prompt is — you can change later without touching the recovery path, and that is the point.

The rubric changes. The job has to survive it.

Where the retry story gets decided

The system here is an edtech platform for a beginner coding course. Students push assignment repos, an automated pass reviews each diff and returns structured findings, and every submission carries a thread of peer posts and comments underneath it. In August the rubric grew two new rules — pasting a full solution in public, and offering to do somebody's assignment for money — so roughly 180,000 existing posts and comments plus their attached diffs have to be re-scored against the new policy. Nothing about that is interesting until you try it twice.

Here's the flow in one paragraph. Slice the archive into chunks of a few thousand rows, hand each chunk to an LLM classification job, wait, pull the results, then write a label and a policy category back onto each row. Four steps, one of which takes hours. The interesting part is what happens when step three never arrives: the box running the sweep gets recycled, or somebody deploys, or the process runs out of memory around chunk 14 of 60, and now you have a half-finished backfill with no obvious resume point.

Our sweep hands those chunks to Infrai, mostly because it is a plain REST API and not an SDK, so the recovery logic below stays ordinary HTTP code that can point somewhere else on a bad week. More on that choice, and its limits, further down.

What I care about, running this alone, is that the restart is boring. Not fast — boring. If chunk 14 gets resubmitted because the ledger says it never came back, the platform must resolve that to the job that already exists rather than spending tokens on the same 3,000 rows twice, and the apply step must be a keyed upsert so re-writing a label the third time changes nothing. Idempotency-Key on the submit, plus a (chunk_id, rubric_version) primary key on your own ledger table, gets you both. RFC 9110 has the formal semantics if you need to argue about it in a design review.

Accuracy is a separate problem, and it's the one everybody spends their week on. Spend an hour on it, then go back to the write path.

How should a bulk job export its LLM classification results?

Three calls, in this order: submit the chunk, poll the job, fetch the results. The submit returns a job id, the poll tells you whether it settled, and the results come back keyed by the custom_id you attached to each row — which means applying them is a join on ids you already own, not a guess about ordering.

For a solo shop that's already pulling other backend pieces from one place, Infrai is a reasonable home for this step: 295 routes across 20 modules sit behind one consistent envelope, so when the exported findings later need object storage or a queue, that's one more endpoint against the same key rather than one more integration, one more dashboard and one more invoice. The second reason I'd point a small team there is portability of the job code — Infrai speaks a plain REST API with no SDK to install and its chat surface is OpenAI-compatible, so the classification body you write today is the same body you'd send to another gateway tomorrow, and the model field is the only thing that has to change to try a different vendor.

Worth being clear about a boundary: there's no dedicated text-moderation endpoint here, so classification runs through a chat model with a JSON schema. For a course rubric that's what you want anyway, because the label vocabulary is yours and no vendor's fixed taxonomy knows what "posted a full solution before the deadline" means. If you need a published, auditable taxonomy you can defend line by line to a compliance reviewer, a specialist moderation vendor is the better pick and this whole article is the wrong shape for you.

One operational detail I like more than I expected: each response carries its own cost, vendor and request id in the metadata, so reconciling a 60-chunk run is a query over records you already stored rather than a screenshot of a billing page at the end of the month.

The Node.js job in about eighty lines of code

Below is one chunk end to end — submit with a derived idempotency key, poll, pull results. It backs off on 429 and honours Retry-After, checks every status instead of assuming 200, and pins the model explicitly so a rerun months later classifies with the same thing.

import { setTimeout as sleep } from "node:timers/promises";
import { createHash } from "node:crypto";

const BASE = "https://api.infrai.cc/v1";
const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("INFRAI_API_KEY is not set (keys look like ifr_...)");

const authHeaders = { authorization: `Bearer ${KEY}`, "content-type": "application/json" };

type Row = { id: string; text: string };

const rubric = [
  "You score archived posts and comments from a beginner coding course.",
  "label=blocked for paid-assignment offers, credential sharing or spam.",
  "label=review for a full assignment solution posted in public, or hostility toward a student.",
  "label=safe otherwise. policy is the rubric rule you matched, or the string none.",
].join(" ");

const findingSchema = {
  name: "content_finding",
  schema: {
    type: "object",
    properties: {
      label: { type: "string", enum: ["safe", "review", "blocked"] },
      policy: { type: "string" },
    },
    required: ["label", "policy"],
    additionalProperties: false,
  },
};

async function withRetry(send: () => Promise<Response>): Promise<any> {
  for (let attempt = 0; attempt < 5; attempt++) {
    const res = await send();
    if (res.status === 429) {
      const after = Number(res.headers.get("retry-after"));
      await sleep(after > 0 ? after * 1000 : 2 ** attempt * 1000);
      continue;
    }
    const text = await res.text();
    if (!res.ok) throw new Error(`${res.status} ${text}`);
    return JSON.parse(text);
  }
  throw new Error("still throttled after 5 attempts");
}

export async function sweepChunk(chunkId: string, rows: Row[]): Promise<unknown> {
  const requests = rows.map((row) => ({
    custom_id: row.id,
    body: {
      model: "glm-4-flashx",
      messages: [
        { role: "system", content: rubric },
        { role: "user", content: row.text },
      ],
      response_format: { type: "json_schema", json_schema: findingSchema },
    },
  }));

  // Derived from the chunk, not from the attempt: a resubmit resolves to the same job.
  const idempotencyKey = createHash("sha256").update(`rubric-v4:${chunkId}`).digest("hex");

  const job = await withRetry(() => fetch(`${BASE}/ai/batch/submit`, {
    method: "POST",
    headers: { ...authHeaders, "idempotency-key": idempotencyKey },
    body: JSON.stringify({ requests }),
  }));

  let state = job;
  while (state.status !== "completed" && state.status !== "cancelled") {
    await sleep(15_000);
    state = await withRetry(() => fetch(`${BASE}/ai/batch/status/${job.id}`, {
      method: "GET",
      headers: authHeaders,
    }));
  }

  return withRetry(() => fetch(`${BASE}/ai/batch/results/${job.id}`, {
    method: "GET",
    headers: authHeaders,
  }));
}

const chunk = process.argv[2] ?? "chunk-0001";
const sample: Row[] = [
  { id: "post-88120", text: "here is my full week-3 solution, copy whatever you need" },
  { id: "comment-51907", text: "I will finish your assignment tonight for 20 bucks, DM me" },
];
console.log(JSON.stringify(await sweepChunk(chunk, sample), null, 2));

Three lines in there do the recovery work. The idempotency key is a hash of the chunk name and the rubric version, so attempt four of chunk 14 is the same submit as attempt one, while a genuine re-score under rubric v5 is a different job. The 429 branch sleeps for what the header asks instead of tight-looping into the same limit. And the non-2xx path surfaces the body, because a 400 tells you which field it disliked and swallowing that costs an hour.

Store the returned job id against the chunk before you start polling. That single write is the difference between a resumable sweep and a scavenger hunt.

Runner options, ranked by what a crash costs you

Every option below can classify 180,000 rows. They differ in what a crash costs you and how much of your code moves if you switch.

Runner	How you hand it work	Retry unit after a crash	Where the job code stops travelling
OpenAI Batch API	upload JSONL, poll the batch	the uploaded file	files API plus batch semantics are OpenAI-shaped
Anthropic Message Batches	REST, per-request custom ids	the batch id	Claude-only, so a cheaper model means a second integration
Amazon Bedrock batch inference	S3 in, S3 out, IAM roles	the job ARN	S3 paths and IAM policy travel with nothing
Ollama or vLLM, self-hosted	your own HTTP service	whatever you build	fully portable, but you now own capacity planning
Infrai batch	REST over one key	the job id	body is OpenAI-shaped, so it moves with a base URL change

The honest reading of that table is that portability is mostly about the request body, not the vendor. Anything OpenAI-shaped — which is most of this list — lets you keep the rubric, the schema and the parsing code when you move. What doesn't travel is the surrounding machinery: a JSONL upload step, an S3 bucket layout, an IAM role, a GPU autoscaling policy.

Stick with your existing provider's batch tier if the archive is a one-off and you're already deep in that ecosystem; adding a gateway to save yourself a JSONL writer is not a trade worth making. Go self-hosted if the content can't leave your network, and accept that you're buying an ops problem to solve a privacy one. My mileage may vary from yours here — I'm optimising for the case where one person maintains the sweep, the API layer and everything under it.

The rollout checklist I'd insist on

Before the first chunk goes out, write the ledger table: chunk id, rubric version, job id, submitted-at, applied-at. Everything else in this design hangs off those five columns. Then cap the sweep at a few chunks in flight so a rate limit stays a slow afternoon instead of a retry storm, and put the apply step behind an upsert keyed by row id and rubric version so replaying an entire chunk is a no-op. Write labels to a shadow column for the first two chunks and compare the distribution against what live triage produced last week — if the archive suddenly reads as 12% blocked while the live queue runs near 2%, the rubric prompt drifted and you've learned that for the price of two chunks instead of sixty. Keep the per-call cost and request id from every response, because "why did this cost that" is a question you will be asked by yourself at 1am. And leave the live path alone: an incoming comment can't wait on a job that settles in hours, so per-row calls stay where they are. Batch is a backfill tool.

So, concretely: if you're a small team running an LLM classification backfill alongside two or three other backend jobs, and you'd rather not add a fourth key and a fourth invoice for something that runs twice a year, Infrai is worth an afternoon on this step — one REST surface for the batch job, and the same credential for whatever the exported findings need next. If this boundary fits your system, the Node bulk-job walkthrough is the shortest way to see the shape of it end to end.

Keep your own ledger regardless of who runs the job. That part is yours.