Process exports with Queues

A big export can take longer than a request should. Instead of holding the HTTP connection open, the API writes a job row, drops a message on a queue, and returns 202 Accepted with a job ID. A consumer picks the message up and does the work.

Let’s create the queue and bind it twice, once as a producer and once as a consumer:

npx wrangler queues create link-vault-exports

{
  "queues": {
    "producers": [{ "binding": "EXPORT_QUEUE", "queue": "link-vault-exports" }],
    "consumers": [{ "queue": "link-vault-exports", "max_retries": 3 }]
  }
}

The producer

The route persists the job first, then sends a small message:

app.post('/api/exports', async c => {
  const jobId = crypto.randomUUID()
  await createExportJob(c.env.DB, { id: jobId, userId, status: 'pending' })
  await c.env.EXPORT_QUEUE.send({ version: 1, jobId, userId, format: 'json' })
  return c.json({ jobId, statusUrl: `/api/exports/${jobId}` }, 202)
})

Notice what’s in the message: IDs, a format, and a schema version. Not the links themselves. Data copied into a message goes stale the moment someone edits a link. The consumer reads fresh data from D1 when it runs.

The consumer

The same Worker exports a queue handler next to fetch:

export default {
  fetch: app.fetch,
  async queue(batch: MessageBatch<ExportMessage>, env: Env) {
    for (const message of batch.messages) {
      try {
        await processExport(env, message.body)
        message.ack()
      } catch (err) {
        console.error(JSON.stringify({ jobId: message.body.jobId, error: String(err) }))
        message.retry()
      }
    }
  }
}

Handle each message on its own. If you throw out of the loop, the whole batch retries, including the messages that already succeeded.

At least once

Queues promise at-least-once delivery. Your consumer may see the same message twice: after a crash, after a timeout, after a retry. So the work must be idempotent, meaning running it twice ends in the same state as running it once.

Derive the R2 key from the job ID, exports/${userId}/${jobId}.json, instead of a random UUID. A retry overwrites the same object rather than creating a second one. Then move the job row through explicit states: pending, running, ready, failed. If the consumer finds the job already ready, it just acks.

Separate failures

A job for a user that no longer exists will never succeed. Mark it failed and ack, or it retries three times for nothing. A D1 timeout is temporary. Retry that one. After max_retries, messages go to a dead_letter_queue if you configure one. Set it, and look at it.

Log the job ID and the message ID. Never log the export contents. Those are someone’s private links.

Now build it end to end: POST /api/exports returns 202 and a job ID, the consumer writes exactly one object to R2, and GET /api/exports/:id reports the status. Then send the same message twice in local dev and confirm there is still only one object in the bucket.