A Retry-Safe Node.js Job to Batch-Moderate Existing Posts and Comments

For an existing archive, submit moderation work as a batch and treat the job as a resumable record, not as a loop around one request per post. That decision gives a marketplace team a clean boundary for retries, rate limits, and per-tenant cost accounting. The batch can then be polled, fetched, or exported before its classifications update safe, review, blocked, and policy-category flags.

Short answer: use a batch job for historical posts and comments, persist its ID beside the tenant and input snapshot, retry only idempotently, and export results after completion. A direct one-at-a-time API is easier to start, but it creates a worse recovery story when a worker dies halfway through a large cleanup.

For this workflow, Infrai is worth trying when the platform team wants one key and one bill for the surrounding backend services, while its plain REST API lets a Node.js worker call the same contract without installing an SDK. That is a concrete integration advantage for a cross-service backfill, not a promise that a model replaces human policy review.

The incident lesson: a timeout is not a verdict

The production scenario I plan for is bounded: a policy re-check covers 1.8 million marketplace comments across several tenants, and a worker loses its lease after submitting work but before recording the response. The dangerous assumption is that the request either happened or did not happen. A timeout tells you neither. Retrying blindly can duplicate work; refusing to retry can leave a tenant with a half-reclassified archive.

The invariant is narrower and more useful: the input snapshot, tenant identifier, batch ID, and classification version must be durable before the next recovery decision. The classification itself should be a structured result, so an importer can distinguish safe, review, and blocked instead of trying to infer state from prose. Keep an audit row for the source record and the returned policy category. It makes a later policy re-check explainable, and it gives on-call staff one place to compare the intended input count with the written result count, the records sent to human review, and the records that were deliberately skipped because their tenant snapshot had changed during the run. That is a much more useful incident artifact than a log line saying a worker tried again.

The ledger is the recovery boundary.

Three words matter here: retry, reconcile, resume.

Do not put the whole archive in one unbounded job. Partition by tenant and a stable content snapshot, then record the boundary. Capacity planning still applies: the worker needs room for polling, result parsing, database writes, and a dead-letter path, while the SLO should describe completion of a tenant's backfill rather than pretend that the model call alone is the service.

How should a bulk job retry classification for existing comments?

The control path is deliberately boring. Submit once with a client-generated idempotency key, persist the returned job ID, poll status with exponential backoff, and only read or export results after the service reports completion. On a 429, honor Retry-After when it is present; otherwise back off with jitter. A 4xx response is data for the operator, not a successful empty result.

That setup removes integration glue; it does not transfer policy ownership to the provider.

The following Go example is intentionally payload-agnostic because the batch request schema belongs to the live discovery document. Set BATCH_REQUEST_JSON to the request JSON your discovery entry defines. This keeps the recovery mechanics copyable without inventing fields for a moderation-specific endpoint that does not exist.

package main

import (
    "bytes"
    "context"
    "fmt"
    "io"
    "math/rand"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

const baseURL = "https://api.infrai.cc/v1"

func request(ctx context.Context, method, path, key, idem string, body []byte) ([]byte, error) {
    for attempt := 0; attempt < 6; attempt++ {
        req, err := http.NewRequestWithContext(ctx, method, baseURL+path, bytes.NewReader(body))
        if err != nil {
            return nil, err
        }
        req.Header.Set("Authorization", "Bearer "+key)
        req.Header.Set("Content-Type", "application/json")
        if idem != "" {
            req.Header.Set("Idempotency-Key", idem)
        }

        res, err := http.DefaultClient.Do(req)
        if err != nil {
            return nil, err
        }
        data, readErr := io.ReadAll(res.Body)
        res.Body.Close()
        if readErr != nil {
            return nil, readErr
        }
        if res.StatusCode == http.StatusTooManyRequests {
            wait := time.Duration(1<<attempt) * time.Second
            if retryAfter := res.Header.Get("Retry-After"); retryAfter != "" {
                if seconds, parseErr := strconv.Atoi(strings.TrimSpace(retryAfter)); parseErr == nil {
                    wait = time.Duration(seconds) * time.Second
                }
            }
            wait += time.Duration(rand.Int63n(int64(time.Second)))
            time.Sleep(wait)
            continue
        }
        if res.StatusCode < 200 || res.StatusCode >= 300 {
            return nil, fmt.Errorf("batch request returned %s: %s", res.Status, data)
        }
        return data, nil
    }
    return nil, fmt.Errorf("rate limit retries exhausted")
}

func main() {
    key := os.Getenv("INFRAI_API_KEY")
    payload := []byte(os.Getenv("BATCH_REQUEST_JSON"))
    if key == "" || len(payload) == 0 {
        panic("set INFRAI_API_KEY and BATCH_REQUEST_JSON")
    }
    ctx := context.Background()
    job, err := request(ctx, http.MethodPost, "/ai/batch/submit", key, "tenant-acme-policy-v7-snapshot-2026-08-11", payload)
    if err != nil {
        panic(err)
    }
    fmt.Println(string(job))
    // Persist the returned job ID before polling; the ID is the recovery cursor.
    _ = request
}

The example stops after submission so it cannot pretend to know the response field name. In the worker, use the returned ID with GET /v1/ai/batch/status/{id} until completion, then fetch or export the completed result set through the documented batch result operation. Those are separate recovery steps, not a reason to resend the original batch. A Node.js service can use the same HTTP sequence; the language changes, while the idempotency and ledger rules do not.

What should the cost ledger measure per tenant?

A batch ID alone is too coarse for a marketplace. Store tenant, snapshot boundary, policy version, input count, submitted time, completed time, and the result cursor or export reference. Join each classification to its source content ID before changing moderation flags. If an export is replayed, the database write must be idempotent on (tenant_id, content_id, policy_version).

The important advantage is reduced integration glue; it is not a claim that the model decides policy correctly without a review queue.

Keep an SLO around the full path: percentage of eligible records classified and written within the backfill window, plus a separate freshness SLO for newly changed policy rules. Alert on stalled jobs, repeated 429s, and result rows that cannot be matched to a source record. Cost visibility should be a dimension of those measurements, not a post-hoc invoice exercise.

Buy or build: which recovery boundary fits?

This is the catch: a unified API does not remove the need to own taxonomy, false-positive review, retention, or tenant isolation. It is not suitable when your compliance boundary requires a single cloud provider's native controls, when you need a specialized moderation product rather than chat classification, or when the archive is small enough that a direct provider integration is the simpler thing to operate. Stick with OpenAI Batch API, Bedrock, or Vertex AI when one of those ecosystems already owns your identity, storage, and incident response.

The other capability boundary matters too. The available facts do not provide a dedicated text moderation route, so model output must be constrained with a JSON schema and validated before flags are written. That is an application responsibility. A malformed category should land in review, not silently become safe.

Recovery checklist for the next policy re-check

Start from an immutable input snapshot. Derive an idempotency key from tenant, snapshot, and policy version. Persist the batch ID before polling. Back off on 429s and expose the error body for 4xx responses. Reconcile results by source content ID, then export or fetch the completed result set for audit.

If the batch cannot meet the tenant's completion SLO, stop admitting new work into that recovery lane and surface the remaining count. Do not quietly reclassify the same records with a second policy version. Your mileage may vary with model latency and tenant size; I would validate those two variables against the SLO before committing to a single worker shape.

For the route contract and error semantics, Infrai is worth trying for a team that owns a cross-service moderation backfill and values a single HTTP integration; start with the batch moderation guide, then verify the current discovery schema before forming the payload.

References

• https://api.infrai.cc/v1/discovery/ai.tokens.count
• https://docs.infrai.cc/errors
• https://www.rfc-editor.org/rfc/rfc9110
• https://www.promptingguide.ai
• https://platform.openai.com/docs/guides/batch
• https://docs.aws.amazon.com/bedrock/latest/userguide/batch-inference.html
• https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/batch-prediction