Building a Durable Document Parsing Pipeline with Cloudflare Queues and Workflows

Flow diagram of a document parsing pipeline built with a Cloudflare Worker, Queue, Workflow, R2, D1, and Mistral OCR — The control path dispatches durable work; D1 and R2 hold product state and large artifacts.

Document extraction looks like one API call in a demo. In production, the hard part is everything around that call: bursts of uploads, PDFs that take minutes to process, provider outages, duplicate delivery, partial batch failures, and users refreshing a progress page while the work is still running.

At PdfParse, we handle those concerns with a Cloudflare-native pipeline. A Worker accepts a job, a Queue absorbs the burst, and a Workflow owns each long-running OCR batch. R2 stores the source documents and raw provider output; D1 stores queryable status and normalized results.

This guide builds that pattern with Cloudflare Workflows and Queues and Mistral's batch OCR endpoint. The example starts after upload: each document already has a stable ID and a short-lived, signed HTTPS URL backed by R2.

The services have different jobs

The request path is:

API Worker → Queue → Workflow → Mistral batch OCR → D1

R2 sits beside the Workflow as durable storage for source PDFs, request JSONL, and raw result JSONL.

The API Worker validates the request, creates a job, splits its documents into batches, and enqueues them.
The Queue levels load and provides at-least-once delivery.
A Workflow instance owns one OCR batch, including submission, polling, result parsing, and persistence.
Mistral batch OCR does the asynchronous document work.
D1 is the product-facing source of truth for progress and extracted fields.
R2 retains the large inputs and outputs that do not belong in a status table.

The key split is between the Queue and the Workflow. The queue consumer should dispatch durable work, acknowledge the message, and finish. The Workflow is the component designed to wait and resume over a longer period.

Define the message contract first

Stable IDs make the pipeline debuggable and safe to replay. An array index is not a document identity: provider output may arrive out of order, and one request can fail while the rest succeed.

// src/contracts.ts
import { z } from "zod";

export const documentInputSchema = z.object({
  id: z.string().min(1),
  url: z.string().url().startsWith("https://"),
});

export const createJobSchema = z.object({
  documents: z.array(documentInputSchema).min(1).max(500),
});

export type DocumentInput = z.infer<typeof documentInputSchema>;

export type ParseBatchMessage = {
  jobId: string;
  batchId: string;
  batchIndex: number;
  totalBatches: number;
  documents: DocumentInput[];
};

The 500-document request cap and batch size used below are application choices, not platform limits. Pick them from measured PDF sizes, page counts, provider quotas, and the amount of work you are willing to retry together.

For persistence, the example assumes three D1 tables:

Table	Unique key	Purpose
`jobs`	`id`	Overall status and total batch count
`batches`	`id`; `workflow_instance_id`	One row per queued OCR batch
`documents`	`(job_id, document_id)`	One terminal result per input document

The uniqueness constraints are the important part. A replay should update the same logical rows instead of appending copies.

Configure the bindings

This wrangler.jsonc runs the API, queue consumer, and Workflow from one Worker. Replace the resource names and D1 ID with your own.

{
  "$schema": "node_modules/wrangler/config-schema.json",
  "name": "document-parser",
  "main": "src/index.ts",
  "compatibility_date": "2026-09-07",
  "compatibility_flags": ["nodejs_compat"],
  "observability": { "enabled": true },

  "queues": {
    "producers": [
      { "binding": "OCR_QUEUE", "queue": "document-ocr" }
    ],
    "consumers": [
      {
        "queue": "document-ocr",
        "max_batch_size": 10,
        "max_batch_timeout": 5,
        "max_retries": 5,
        "dead_letter_queue": "document-ocr-dlq"
      }
    ]
  },

  "workflows": [
    {
      "binding": "OCR_WORKFLOW",
      "name": "document-ocr-workflow",
      "class_name": "OcrWorkflow"
    }
  ],

  "d1_databases": [
    {
      "binding": "DB",
      "database_name": "document-parser",
      "database_id": "YOUR_D1_DATABASE_ID"
    }
  ],

  "r2_buckets": [
    { "binding": "DOCUMENTS", "bucket_name": "document-parser" }
  ]
}

Install the runtime dependencies, generate types from the bindings, and add the provider key as a secret:

pnpm add @mistralai/mistralai zod
pnpm add -D wrangler typescript
pnpm wrangler types
pnpm wrangler secret put MISTRAL_API_KEY

Regenerating worker-configuration.d.ts after a binding change keeps the TypeScript environment aligned with the deployed configuration.

Accept a job and enqueue bounded batches

The API writes the job before publishing messages. If publishing fails, the job remains visible and can be reconciled instead of disappearing.

// src/index.ts
import {
  createJobSchema,
  type ParseBatchMessage,
} from "./contracts";

const DOCUMENTS_PER_BATCH = 50;

function chunk<T>(items: T[], size: number): T[][] {
  return Array.from(
    { length: Math.ceil(items.length / size) },
    (_, index) => items.slice(index * size, (index + 1) * size),
  );
}

async function createJob(request: Request, env: Env): Promise<Response> {
  const parsed = createJobSchema.safeParse(await request.json());
  if (!parsed.success) {
    return Response.json(
      { error: "Invalid request", issues: parsed.error.issues },
      { status: 400 },
    );
  }

  const jobId = crypto.randomUUID();
  const batches = chunk(parsed.data.documents, DOCUMENTS_PER_BATCH);

  await env.DB.prepare(
    `INSERT INTO jobs (id, status, total_batches, completed_batches)
     VALUES (?, 'queued', ?, 0)`,
  ).bind(jobId, batches.length).run();

  const messages = batches.map((documents, batchIndex) => {
    const body: ParseBatchMessage = {
      jobId,
      batchId: `${jobId}:${batchIndex}`,
      batchIndex,
      totalBatches: batches.length,
      documents,
    };
    return { body };
  });

  await env.OCR_QUEUE.sendBatch(messages);
  return Response.json({ jobId, batches: batches.length }, { status: 202 });
}

This sample accepts URLs to keep the pipeline code focused. In a real API, the server should resolve object keys to signed R2 URLs after checking ownership. Passing arbitrary user-supplied remote URLs directly to an OCR provider would create an SSRF risk.

Dispatch one Workflow per logical batch

Cloudflare Queues uses at-least-once delivery, so the same message can arrive more than once. Deriving the Workflow instance ID from batchId turns that duplicate delivery into a lookup rather than duplicate OCR work.

async function dispatchBatch(
  message: Message<ParseBatchMessage>,
  env: Env,
): Promise<void> {
  const instanceId = `ocr-${message.body.batchId}`;

  try {
    const existing = await env.OCR_WORKFLOW.get(instanceId);
    const { status } = await existing.status();

    if (status === "unknown") {
      await env.OCR_WORKFLOW.create({
        id: instanceId,
        params: message.body,
      });
    }

    message.ack();
  } catch (error) {
    console.error("workflow dispatch failed", {
      batchId: message.body.batchId,
      error: error instanceof Error ? error.message : String(error),
    });
    message.retry({ delaySeconds: 10 });
  }
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    if (request.method === "POST" && url.pathname === "/jobs") {
      return createJob(request, env);
    }
    return new Response("Not found", { status: 404 });
  },

  async queue(
    batch: MessageBatch<ParseBatchMessage>,
    env: Env,
  ): Promise<void> {
    await Promise.all(
      batch.messages.map((message) => dispatchBatch(message, env)),
    );
  },
} satisfies ExportedHandler<Env, ParseBatchMessage>;

export { OcrWorkflow } from "./workflow";

There is a small race worth understanding: two deliveries may both observe unknown. One create() succeeds and the other sees an existing-ID error. The second message retries, finds the instance, and acknowledges. The deterministic ID is what makes the race converge.

Build the OCR batch in Mistral's actual format

Mistral's batch API accepts a JSONL file with one request object per line. The batch job declares endpoint: "/v1/ocr"; each line contains the OCR request body. Structured extraction is specified with document_annotation_format.

// src/mistral.ts
import { Mistral } from "@mistralai/mistralai";
import { z } from "zod";
import type { DocumentInput } from "./contracts";

export const invoiceSchema = z.object({
  invoice_number: z.string().nullable(),
  vendor_name: z.string().nullable(),
  invoice_date: z.string().nullable(),
  currency: z.string().length(3).nullable(),
  total_amount: z.number().nonnegative().nullable(),
});

export type Invoice = z.infer<typeof invoiceSchema>;

export async function submitOcrBatch(
  apiKey: string,
  batchId: string,
  documents: DocumentInput[],
): Promise<{ providerBatchId: string; inputFileId: string }> {
  const client = new Mistral({ apiKey });
  const schema = z.toJSONSchema(invoiceSchema, { target: "draft-7" });

  const jsonl = documents.map((document) => JSON.stringify({
    custom_id: document.id,
    body: {
      model: "mistral-ocr-latest",
      document: {
        type: "document_url",
        document_url: document.url,
      },
      document_annotation_format: {
        type: "json_schema",
        json_schema: { name: "Invoice", schema, strict: true },
      },
    },
  })).join("\n");

  const input = new File(
    [jsonl],
    `${batchId}.jsonl`,
    { type: "application/x-ndjson" },
  );
  const uploaded = await client.files.upload({ file: input, purpose: "batch" });
  const batch = await client.batch.jobs.create({
    inputFiles: [uploaded.id],
    model: "mistral-ocr-latest",
    endpoint: "/v1/ocr",
    metadata: { batchId },
  });

  return { providerBatchId: batch.id, inputFileId: uploaded.id };
}

custom_id is the join key between the input and output. The invoice number extracted from the PDF is business data, not a reliable database identity.

Poll with durable sleeps

A Workflow can make one status request, persist the result of that step, and then sleep if the provider is still running. This exposes useful checkpoints and avoids keeping a callback open while nothing is happening.

Flow diagram showing duplicate queue delivery, deterministic Workflow creation, durable polling, and idempotent result writes — Transport may repeat. Stable IDs and idempotent writes make each replay converge on the same product result.

// src/workflow.ts
import {
  NonRetryableError,
  WorkflowEntrypoint,
  type WorkflowEvent,
  type WorkflowStep,
} from "cloudflare:workers";
import { Mistral } from "@mistralai/mistralai";
import type { ParseBatchMessage } from "./contracts";
import { downloadResults, submitOcrBatch } from "./mistral";
import { saveResults } from "./persistence";

const TERMINAL_FAILURES = new Set([
  "FAILED",
  "CANCELLED",
  "TIMEOUT_EXCEEDED",
]);

export class OcrWorkflow extends WorkflowEntrypoint<Env, ParseBatchMessage> {
  async run(
    event: WorkflowEvent<ParseBatchMessage>,
    step: WorkflowStep,
  ): Promise<{ processed: number }> {
    const input = event.payload;

    await step.do("mark batch running", async () => {
      await this.env.DB.prepare(
        `INSERT INTO batches
           (id, job_id, workflow_instance_id, status, document_count)
         VALUES (?, ?, ?, 'running', ?)
         ON CONFLICT(id) DO UPDATE SET status = 'running'`,
      ).bind(
        input.batchId,
        input.jobId,
        event.instanceId,
        input.documents.length,
      ).run();
    });

    const submitted = await step.do(
      "submit OCR batch",
      { retries: { limit: 1, delay: "1 second", backoff: "constant" } },
      async () => submitOcrBatch(
        this.env.MISTRAL_API_KEY,
        input.batchId,
        [...input.documents],
      ),
    );

    const client = new Mistral({ apiKey: this.env.MISTRAL_API_KEY });
    let outputFileId: string | undefined;

    for (let check = 1; check <= 120; check += 1) {
      const batch = await step.do(
        `check OCR batch ${check}`,
        {
          retries: {
            limit: 3,
            delay: "5 seconds",
            backoff: "exponential",
          },
          timeout: "30 seconds",
        },
        async () => {
          const current = await client.batch.jobs.get({
            jobId: submitted.providerBatchId,
          });
          return {
            status: current.status,
            outputFile: current.outputFile ?? null,
          };
        },
      );

      if (batch.status === "SUCCESS") {
        if (!batch.outputFile) {
          throw new NonRetryableError("Successful batch has no output file");
        }
        outputFileId = batch.outputFile;
        break;
      }

      if (TERMINAL_FAILURES.has(batch.status)) {
        throw new NonRetryableError(`Mistral batch ended as ${batch.status}`);
      }

      await step.sleep(`wait before check ${check + 1}`, "30 seconds");
    }

    if (!outputFileId) {
      throw new NonRetryableError("Mistral batch exceeded the polling window");
    }

    const results = await step.do(
      "download and validate results",
      {
        retries: {
          limit: 5,
          delay: "5 seconds",
          backoff: "exponential",
        },
        timeout: "2 minutes",
      },
      async () => downloadResults(client, outputFileId),
    );

    await step.do("commit results", async () => {
      await saveResults(this.env.DB, input.jobId, input.batchId, results);
    });

    return { processed: results.length };
  }
}

At 30-second intervals, 120 checks give this Workflow a one-hour polling window. Tune that window from observed batch duration and the lifetime of the signed document URLs.

The submission step uses one attempt for a reason. A provider may accept the request and then drop the connection before returning the new batch ID. Retrying immediately could create two provider batches. A production reconciliation path can search provider metadata for batchId before an operator resubmits.

Validate every output line

A successful provider batch can still contain failed documents. Parse the output JSONL line by line, inspect the embedded HTTP status, and validate document_annotation with the same Zod schema used for the request.

const batchOutputSchema = z.object({
  custom_id: z.string(),
  response: z.object({
    status_code: z.number().int(),
    body: z.object({
      document_annotation: z.string().optional(),
      message: z.string().optional(),
      detail: z.string().optional(),
    }),
  }).optional(),
  error: z.object({ message: z.string() }).optional(),
});

export type DocumentResult =
  | { documentId: string; status: "completed"; invoice: Invoice }
  | { documentId: string; status: "failed"; error: string };

export async function downloadResults(
  client: Mistral,
  fileId: string,
): Promise<DocumentResult[]> {
  const stream = await client.files.download({ fileId });
  const text = await new Response(stream).text();

  return text.split("\n").filter(Boolean).map((line) => {
    const item = batchOutputSchema.parse(JSON.parse(line));

    if (item.error) {
      return {
        documentId: item.custom_id,
        status: "failed",
        error: item.error.message,
      };
    }

    const response = item.response;
    if (!response || response.status_code < 200 || response.status_code >= 300) {
      return {
        documentId: item.custom_id,
        status: "failed",
        error: response?.body.message ?? response?.body.detail ?? "OCR request failed",
      };
    }

    const annotation = response.body.document_annotation;
    if (!annotation) {
      return {
        documentId: item.custom_id,
        status: "failed",
        error: "Missing document annotation",
      };
    }

    return {
      documentId: item.custom_id,
      status: "completed",
      invoice: invoiceSchema.parse(JSON.parse(annotation)),
    };
  });
}

Representing an individual document failure as data allows the other results to commit. Throwing still makes sense when the batch itself cannot be interpreted—for example, if the downloaded file is not valid JSONL.

Make the database commit safe to replay

Workflow steps may retry, operators may restart an errored instance, and deployments happen while jobs are running. The database write should converge on the same state each time it runs.

// src/persistence.ts
import type { DocumentResult } from "./mistral";

export async function saveResults(
  db: D1Database,
  jobId: string,
  batchId: string,
  results: DocumentResult[],
): Promise<void> {
  const statements = results.map((result) => db.prepare(
    `INSERT INTO documents
       (job_id, document_id, batch_id, status, result_json, error)
     VALUES (?, ?, ?, ?, ?, ?)
     ON CONFLICT(job_id, document_id) DO UPDATE SET
       batch_id = excluded.batch_id,
       status = excluded.status,
       result_json = excluded.result_json,
       error = excluded.error`,
  ).bind(
    jobId,
    result.documentId,
    batchId,
    result.status,
    result.status === "completed" ? JSON.stringify(result.invoice) : null,
    result.status === "failed" ? result.error : null,
  ));

  statements.push(
    db.prepare("UPDATE batches SET status = 'completed' WHERE id = ?")
      .bind(batchId),
  );

  await db.batch(statements);

  // Deriving progress from terminal batch rows avoids double increments.
  await db.prepare(
    `UPDATE jobs
     SET completed_batches = (
       SELECT COUNT(*) FROM batches
       WHERE job_id = ? AND status IN ('completed', 'failed')
     ),
     status = CASE
       WHEN total_batches = (
         SELECT COUNT(*) FROM batches
         WHERE job_id = ? AND status IN ('completed', 'failed')
       ) THEN 'completed'
       ELSE 'processing'
     END
     WHERE id = ?`,
  ).bind(jobId, jobId, jobId).run();
}

For larger result sets, write bounded groups rather than building an arbitrarily large D1 batch.

The actual delivery guarantee

This architecture does not make every distributed component execute exactly once. It combines:

at-least-once queue delivery;
one intended Workflow instance per deterministic batch ID;
durable Workflow step results and sleeps;
idempotent database commits; and
per-document terminal outcomes.

The useful product guarantee is: replaying the pipeline converges on one result per document.

That statement leaves room for the real external-write edge case described above while still giving developers and operators a system they can retry confidently.

Test the failure paths

The happy path is the least interesting test. Before deploying, exercise these cases:

Deliver the same queue message twice. Only one Workflow should continue.
Return RUNNING for several checks. The Workflow should sleep and resume with earlier step outputs intact.
Fail one JSONL line. The other documents should still commit.
Throw during the result commit, then replay it. Counts and rows should not duplicate.
Exhaust queue retries. The message should arrive in a monitored dead-letter queue.
Expire a signed R2 URL. The provider failure should remain attributable to one document.
Send an oversized or unauthorized request. It should fail before any work is queued.

Log the same jobId, batchId, Workflow instance ID, provider batch ID, and document ID at their respective boundaries. Those identifiers turn “OCR is stuck” into a traceable state transition.

Why this shape scales

Scaling this system is less about maximizing concurrency than containing the cost of failure.

The API stays responsive because it validates, records, and enqueues. Queue settings control pressure on downstream services. Each Workflow checkpoints the expensive lifecycle. A bad PDF becomes one failed document rather than a poisoned 500-document job. Idempotent writes make recovery routine.

That is the pattern: Queues absorb demand, Workflows own time, R2 holds bytes, D1 exposes state, and stable IDs make the boundaries safe to repeat.