A Background Job Queue with Live Progress: Nuxt 4 + TypeScript + Nitro Tasks + Redis + SSE
Somebody clicks "Export" and your handler starts building a 40,000-row report. Ninety seconds later one of four things has happened: the browser gave up, your reverse proxy gave up, the platform's 60-second function limit killed it, or the user hit refresh and started a second one. The spinner told them nothing the whole time.
The instinct is to make the export faster. The fix is to stop returning it from the request at all. The request's job is to accept the work — twenty milliseconds, 202, here's an id — and something else does the work while a separate connection narrates it.
That normally means adding a service: a queue broker, a worker process, a job library, a dashboard. What makes this build interesting is that Nitro already has every piece. A task runner with a scheduler, a pluggable KV layer, and streaming responses are all in the box. You're wiring together things you already deploy.
Tags: Nuxt 4, Nitro Tasks, Redis, unstorage, Server-Sent Events, TypeScript
Time to read: 19 min
What you'll build: an export queue. Click a button, get a job id back immediately, watch a progress bar move in real time as a server-side task grinds through the work, and get a link at the end. Jobs survive a server restart, stuck jobs get reaped on a cron, and the whole thing runs on one Node process plus a Redis you probably already have.
Why this combination
Four pieces, and the interesting part is where the seams are.
Nitro Tasks are the unit of background work. defineTask in server/tasks/, and Nitro gives you a name, a payload, a CLI, a dev-server endpoint, and — the part that matters here — a cron scheduler that fires them. They're marked experimental and have been for a while; that's a real caveat, and I'll come back to what it costs you.
unstorage with the Redis driver is where jobs live. Nitro's useStorage() is a key-value API with a driver underneath, which means the queue is fs on your laptop and Redis in production with a config change and no code change. It also means job state survives a deploy, which an in-memory Map does not.
Server-Sent Events carry progress to the browser. Progress is one-directional: the server talks, the client listens. SSE is a text/event-stream response, has automatic reconnection built into EventSource, needs no protocol upgrade, and — unlike a WebSocket — survives every corporate proxy in existence. If you need the client to talk back, you want Nitro's WebSocket support instead; this is the other half of that story.
TypeScript and shared/ are what stop this becoming three loosely-related pieces of code. There is one Job interface. The task writes it, the SSE endpoint serialises it, the component renders it, and the compiler checks all three against the same declaration.
What you're deliberately not getting: BullMQ's retry backoff, priorities, rate limiting, and dead-letter queues. If you need those, use BullMQ. This is the tier below — the one that covers "we have three slow endpoints and no infrastructure budget."
Prerequisites
- Node.js 20+
- Nuxt 4.x (this was written against 4.2 / Nitro 2.12)
- A Redis you can reach.
docker run -p 6379:6379 redis:7-alpineis enough - Comfort with
server/in Nuxt — handlers,useStorage,defineEventHandler
1) Scaffold and configure
npx nuxi@latest init nuxt4-jobs
cd nuxt4-jobs
npm i ioredis
ioredis is the only runtime dependency. unstorage ships the Redis driver but not the client — the driver imports ioredis and expects you to have installed it, which is why a fresh project fails with a module-not-found the first time you touch the mount.
Everything else is config:
// nuxt.config.ts
export default defineNuxtConfig({
compatibilityDate: '2025-07-15',
nitro: {
experimental: {
// Without this flag, server/tasks/ is never scanned. The tasks
// exist as files and are invisible to the runtime.
tasks: true,
},
storage: {
jobs: {
driver: 'redis',
url: process.env.REDIS_URL,
// Namespaces every key as `jobs:*` inside Redis, so this
// can share a database with your cache without colliding.
base: 'jobs',
},
},
devStorage: {
// Local dev writes JSON files under .data/jobs instead.
// Same code, no Redis required to run `npm run dev`.
jobs: { driver: 'fs', base: './.data/jobs' },
},
scheduledTasks: {
// The safety net: if a job was enqueued and nothing picked
// it up, this catches it within a minute.
'* * * * *': ['jobs:drain'],
// Housekeeping, hourly.
'0 * * * *': ['jobs:reap'],
},
},
})
Three things worth knowing before you move on.
experimental.tasks is not optional and not loud. Leave it out and server/tasks/ is simply never scanned. runTask still exists — it's auto-imported unconditionally — so what you get is a runtime 404 reading Task `jobs:drain` is not available! from a call site that looks perfectly correct. The error never mentions the config.
devStorage is the reason this is pleasant to work on. The fs driver writes one file per key under .data/jobs, which means debugging a stuck job is cat .data/jobs/job/abc123. Note the missing extension — the driver maps : to / and writes the value verbatim, so the file is JSON with no .json on the end. Add .data to .gitignore.
scheduledTasks is per-preset, and the support matrix is thinner than it looks. The node-server, bun, deno and dev presets run an in-process croner scheduler, which is the case this article is written for. Cloudflare wires its native scheduled() handler to your scheduledTasks map, but only dispatches crons you have declared yourself under [triggers] in wrangler.toml — and the expression has to match the key exactly. Vercel and Netlify have no wiring at all in Nitro 2.12; the config is accepted and nothing fires. If your deploy target is serverless, read section 11 before you build on this.
2) One shape, in shared/
The Vue app and the Nitro server both need to know what a job is, and shared/ is the one directory both can import from:
// shared/jobs.ts
export type JobStatus = 'queued' | 'running' | 'done' | 'failed'
export interface Job {
id: string
status: JobStatus
/** 0..1. The only number the progress bar cares about. */
progress: number
/** Human-readable stage, e.g. "Rendering rows 12000–13000". */
message: string
input: { rows: number }
attempts: number
createdAt: number
startedAt?: number
finishedAt?: number
/** Set when status is 'done'. */
result?: { url: string, bytes: number }
/** Set when status is 'failed'. Safe to show a user. */
error?: string
/**
* Incremented on every write. This is the change-detector the SSE
* endpoint compares against, and it doubles as the SSE event id.
*/
version: number
}
export const TERMINAL: JobStatus[] = ['done', 'failed']
export function isTerminal(job: Job | null): boolean {
return !!job && TERMINAL.includes(job.status)
}
version is the one field that isn't obvious, and it earns its place twice. The SSE endpoint in section 7 polls the job record and needs to know whether anything actually changed — comparing one integer beats deep-equalling an object on every tick. And SSE messages carry an id that the browser echoes back as Last-Event-ID when it reconnects, so a monotonic counter is exactly the right value to put there.
Note also what's not in Job: no Date objects, no functions, no class instances. This goes through JSON.stringify at least twice on its way to the browser. Timestamps are numbers.
3) The queue, on top of key-value
unstorage gives you getItem, setItem, getKeys. A queue needs a list and a claim. Both are buildable, and one of them has a caveat you need to hear before you use it:
// server/utils/queue.ts
import type { Job } from '#shared/jobs'
const QUEUE_KEY = 'queue'
const jobs = () => useStorage<Job>('jobs')
// A second, untyped handle: the queue is a string[], not a Job.
const meta = () => useStorage<string[]>('jobs')
function key(id: string) {
// In unstorage, ':' is the path separator — this is a nested key,
// and it maps straight onto Redis key namespacing.
return `job:${id}`
}
export async function createJob(input: Job['input']): Promise<Job> {
const job: Job = {
id: crypto.randomUUID(),
status: 'queued',
progress: 0,
message: 'Queued',
input,
attempts: 0,
createdAt: Date.now(),
version: 1,
}
await jobs().setItem(key(job.id), job)
const pending = (await meta().getItem(QUEUE_KEY)) ?? []
pending.push(job.id)
await meta().setItem(QUEUE_KEY, pending)
return job
}
export async function readJob(id: string): Promise<Job | null> {
return await jobs().getItem(key(id))
}
/** Every mutation goes through here, so `version` can never be forgotten. */
export async function patchJob(id: string, patch: Partial<Job>): Promise<Job | null> {
const current = await readJob(id)
if (!current) return null
const next: Job = { ...current, ...patch, version: current.version + 1 }
await jobs().setItem(key(id), next)
return next
}
/** Pop the next queued job and mark it running. Returns null when empty. */
export async function claimNext(): Promise<Job | null> {
const pending = (await meta().getItem(QUEUE_KEY)) ?? []
try {
let id: string | undefined
// Skip ids whose record has been reaped out from under the queue.
// Returning null on the first orphan would end the drain early
// while real work was still waiting behind it.
while ((id = pending.shift())) {
const claimed = await patchJob(id, {
status: 'running',
startedAt: Date.now(),
message: 'Starting',
})
if (claimed) return claimed
}
return null
}
finally {
// Whatever we consumed comes off the queue either way.
await meta().setItem(QUEUE_KEY, pending)
}
}
Now the caveat, because pretending it isn't there would be doing you a disservice.
claimNext is a read-modify-write, and unstorage has no atomic primitives. Two drains running at the same instant can both read the same array, both shift() the same id, and both process the job. Within a single Node process that never happens — section 5 explains why — but two PM2 cluster workers or two containers behind a load balancer will do it eventually.
If that's your deployment, you need a real atomic pop, and the Redis driver leaves you a door:
// server/utils/redis.ts
import type Redis from 'ioredis'
/** The live ioredis client behind the `jobs` mount, or null on fs/memory. */
export function redisClient(): Redis | null {
const { driver } = useStorage().getMount('jobs')
return (driver.getInstance?.() as Redis | undefined) ?? null
}
With that, claimNext becomes an LPOP — one round trip, atomic by definition, and the array-in-a-key version stays as the fallback for local fs development. That escape hatch is the honest answer to "is unstorage enough?": it's enough until you have more than one writer, and then it hands you the client.
4) The task that does the work
server/tasks/ maps filenames to task names, joining directories with :. So server/tasks/jobs/drain.ts is the task jobs:drain.
Everything in server/utils/ is auto-imported across the whole Nitro context — handlers, plugins and tasks alike — so claimNext, patchJob and buildReport need no import statements below. (Resist the urge to reach for a #server/... alias here: that landed in Nuxt 4.3, and on 4.2 it simply fails to resolve.)
// server/tasks/jobs/drain.ts
/** Stop claiming new work before the next cron tick lands. */
const BUDGET_MS = 50_000
export default defineTask({
meta: {
name: 'jobs:drain',
description: 'Process queued export jobs until the queue is empty',
},
async run() {
const deadline = Date.now() + BUDGET_MS
let processed = 0
while (Date.now() < deadline) {
const job = await claimNext()
if (!job) break
try {
const result = await buildReport(job)
await patchJob(job.id, {
status: 'done',
progress: 1,
message: 'Complete',
finishedAt: Date.now(),
result,
})
}
catch (error) {
// The message is shown to the user, so keep it a string
// and keep the stack in the server log where it belongs.
console.error(`[jobs:drain] ${job.id} failed`, error)
await patchJob(job.id, {
status: 'failed',
message: 'Failed',
finishedAt: Date.now(),
error: error instanceof Error ? error.message : 'Unknown error',
})
}
processed++
}
return { result: { processed } }
},
})
The while loop with a time budget is doing something specific. A drain that handles exactly one job per cron tick processes 60 jobs an hour, which is useless. A drain with no budget at all is still running when the next tick fires. Fifty seconds against a one-minute cron leaves headroom for the job in flight to finish and for the loop to exit cleanly.
And the work itself, kept in its own file because it's the only part you'd swap for something real:
// server/utils/report.ts
import type { Job } from '#shared/jobs'
import { patchJob } from './queue'
const CHUNK = 500
export async function buildReport(job: Job): Promise<NonNullable<Job['result']>> {
const total = job.input.rows
let bytes = 0
for (let row = 0; row < total; row += CHUNK) {
const end = Math.min(row + CHUNK, total)
// Stand-in for the expensive thing: a query, a PDF render, an
// image resize, a third-party API you're rate-limited against.
await new Promise(resolve => setTimeout(resolve, 120))
bytes += (end - row) * 96
await patchJob(job.id, {
progress: end / total,
message: `Rendering rows ${row.toLocaleString()}–${end.toLocaleString()}`,
})
}
return { url: `/exports/${job.id}.csv`, bytes }
}
Progress granularity is a write to Redis, so chunk it. One patchJob per row on a 40,000-row export is 40,000 round trips and a progress bar that updates faster than a screen refreshes. A chunk size that produces somewhere between 20 and 200 updates over the life of the job is the range where the bar looks alive and the storage load is noise.
5) Enqueue, and the dedupe that makes it safe
The endpoint that accepts work does three things and none of them are slow:
// server/api/jobs.post.ts
export default defineEventHandler(async (event) => {
const body = await readBody<{ rows?: unknown }>(event)
const rows = Number(body?.rows)
if (!Number.isInteger(rows) || rows < 1 || rows > 200_000) {
throw createError({ statusCode: 400, statusMessage: 'rows must be 1..200000' })
}
const job = await createJob({ rows })
// Kick the drain now rather than waiting up to 60s for the cron.
// The promise is deliberately not awaited — see below.
event.waitUntil(
runTask('jobs:drain').catch(error =>
console.error('[jobs] drain kick failed', error),
),
)
setResponseStatus(event, 202)
return { id: job.id }
})
That runTask call looks reckless. Every enqueue kicks a drain, so ten users clicking Export at once fire ten drains, each running a while loop over a shared queue — exactly the race from section 3.
It isn't, and the reason is a detail of Nitro's implementation worth internalising:
Each task can have one running instance. Calling a task of the same name multiple times in parallel results in calling it once, and all callers get the same return value.
Nitro keeps a map of in-flight tasks keyed by name. The second runTask('jobs:drain') doesn't start a second drain; it returns the promise the first one is already running. Ten enqueues produce one drain loop that happens to find ten jobs waiting.
This is the load-bearing idea of the whole architecture, so let's be precise about what it does and doesn't buy you. It means one drain per process, which is what makes the non-atomic claimNext safe on a single-instance deploy. It does not mean one drain per cluster — the map is a module-level object, so two Node processes have two of them. And it means your tasks must be named per worker, not per job: there is no runTask('job:' + id) design here, because you'd get one instance per id and lose the serialisation entirely.
event.waitUntil is the other half, and it's worth being precise about what it actually does in Nitro 2, because the name promises more than it delivers. It records the promise on the event and forwards it to the platform's native waitUntil if the runtime provides one — which on Cloudflare is the difference between your drain finishing and the isolate being frozen the instant the response flushes. On the Node preset there is no native hook, so nothing awaits it: the drain runs purely because the promise has already started, and a SIGTERM mid-drain will cut it off.
That isn't a reason to drop the call — it's the correct API and it becomes load-bearing the day you deploy to an edge runtime — but on Node it's the reaper in section 10, not waitUntil, that makes a killed drain recoverable.
6) Why progress is polled, not pushed
The task writes progress into Redis. The SSE endpoint has to notice. There are three ways to bridge that gap and it's worth knowing why the boring one wins.
An in-process EventEmitter in a Nitro plugin is the fastest and simplest — until the drain runs in worker 2 and the SSE connection landed on worker 1, at which point the browser watches a job it will never hear about.
Redis pub/sub is the correct answer for multiple processes, and redisClient() from section 3 gets you there in about fifteen lines. It's also a second connection per subscriber, a second failure mode, and a message-ordering question you now own.
Polling the job record costs one GET per connection per tick. At 250ms and fifty concurrent watchers that's 200 reads per second against Redis, which is roughly nothing — Redis does six figures. It works identically across one process or twenty. It has no ordering problem, because you always read the latest state rather than replaying a sequence of deltas.
So: poll, ship it, and reach for pub/sub when you have thousands of concurrent watchers rather than dozens. The version field is what makes the poll cheap on the wire — you read every 250ms but only send when the integer moved.
Polling also gives you frame-coalescing for free, which is worth understanding rather than mistaking for a bug. If the task writes three progress updates inside one 250ms window, the client receives one message carrying the newest of them; you'll watch version numbers skip in the stream. That's the correct behaviour for a progress bar — it should show where the job is, not replay where it has been — and it's the reason a chatty task can't flood a slow connection.
7) The SSE endpoint
// server/api/jobs/[id]/events.get.ts
import { isTerminal } from '#shared/jobs'
const TICK_MS = 250
export default defineEventHandler(async (event) => {
const id = getRouterParam(event, 'id')!
const initial = await readJob(id)
if (!initial) {
// Throw *before* creating the stream — once headers are sent as
// text/event-stream you can no longer return a 404.
throw createError({ statusCode: 404, statusMessage: 'No such job' })
}
const stream = createEventStream(event)
let lastVersion = -1
let timer: ReturnType<typeof setInterval> | undefined
async function tick() {
const job = await readJob(id)
if (!job) {
await stream.push({ event: 'gone', data: id })
return stop()
}
if (job.version !== lastVersion) {
lastVersion = job.version
await stream.push({
id: String(job.version),
event: 'progress',
data: JSON.stringify(job),
})
}
if (isTerminal(job)) {
await stream.push({ event: 'end', data: job.status })
return stop()
}
}
function stop() {
clearInterval(timer)
return stream.close()
}
// Fires when the stream closes for any reason — client disconnect,
// closed tab, dropped wifi, or our own stop(). Without this the
// interval outlives the connection, once per abandoned tab.
stream.onClosed(() => clearInterval(timer))
// Nothing here is awaited, deliberately. See below.
void stream.push({ retry: 3000, event: 'ready', data: String(initial.version) })
timer = setInterval(() => void tick(), TICK_MS)
void tick()
return stream.send()
})
Five things in there are the difference between this working and this being subtly broken.
Never await a push() before send(). This is the one that will cost you an afternoon. push() writes into a TransformStream whose readable side has nothing attached to it until send() hands it to the response — and a TransformStream applies backpressure from byte zero. Await that first write and the promise never settles, send() is never reached, no headers are ever flushed, and the browser sits on a request that will hang until it times out. It looks exactly like a Redis problem and it is not. Fire the setup pushes with void, then return send().
stream.send() sets the headers you'd otherwise get wrong — and it sets them at send() time, not at construction. Content-Type: text/event-stream, a Cache-Control that defeats every intermediate cache, Connection: keep-alive on HTTP/1.1 only, and X-Accel-Buffering: no, which is the header Nginx reads to disable response buffering for this one response. That last one is the single most common SSE deployment bug, and h3 has already fixed it for you. Section 11 covers what's left.
return stream.send() must be the last thing. The returned promise resolves when the stream ends; returning it is what tells h3 to hand the response body over rather than serialising a return value. Register your interval and listeners first — after send() the handler has already yielded control.
Named events aren't decoration. event: 'progress' and event: 'end' become distinct addEventListener targets on the client. Without a name everything arrives as the default message event and you're branching on the payload to work out what you got.
retry: 3000 sets the client's reconnection delay. EventSource reconnects on its own; the default gap is browser-dependent — 3 seconds in Chromium, 5 in Firefox. Sending retry makes it a number you chose, and it's the knob you'd turn to back off when the queue is under load.
8) The client half
EventSource is a browser API, so the composable has to be careful about when it runs:
// app/composables/useJob.ts
import type { Job } from '#shared/jobs'
export function useJob() {
const job = shallowRef<Job | null>(null)
const error = ref<string | null>(null)
const connected = ref(false)
let source: EventSource | null = null
function stop() {
source?.close()
source = null
connected.value = false
}
function watchJob(id: string) {
// Never construct EventSource during SSR — it doesn't exist in Node.
if (import.meta.server) return
stop()
const es = new EventSource(`/api/jobs/${id}/events`)
source = es
es.addEventListener('open', () => {
connected.value = true
error.value = null
})
es.addEventListener('progress', (message) => {
job.value = JSON.parse((message as MessageEvent).data) as Job
})
// The server finished and closed. THIS is what stops the browser
// from reconnecting — see below.
es.addEventListener('end', () => stop())
es.addEventListener('gone', () => {
error.value = 'That job no longer exists'
stop()
})
es.onerror = () => {
connected.value = false
// Do not close here: readyState CONNECTING means the browser is
// already retrying, and closing would cancel a recovery that
// was about to succeed.
if (es.readyState === EventSource.CLOSED)
error.value = 'Lost connection to the job stream'
}
}
async function start(rows: number) {
const { id } = await $fetch<{ id: string }>('/api/jobs', {
method: 'POST',
body: { rows },
})
watchJob(id)
return id
}
onScopeDispose(stop)
return { job, error, connected, start, watchJob, stop }
}
The end listener is the piece everybody gets wrong the first time, and it's worth stating plainly:
EventSource reconnects after every disconnection, including a clean one. There is no in-band "we're done here" frame in the protocol. Close the stream from the server and the browser waits retry milliseconds and connects again — forever, in a loop, hitting an endpoint for a job that finished ten minutes ago. Two ways out: the client calls close(), which is why the server sends an explicit end event whose entire purpose is to trigger that call, or the server answers the reconnect with 204 No Content, which permanently stops the retry loop per spec. Do the first; the second is a good belt-and-braces addition to the endpoint if you have clients you don't control.
Two smaller things. shallowRef rather than ref, because a whole new Job object arrives on every message — deep reactivity would walk every property of a value you're replacing wholesale. And onScopeDispose rather than onUnmounted, so the connection also closes when the composable is used inside an effect scope that gets stopped.
9) The component
<!-- app/pages/index.vue -->
<script setup lang="ts">
import { isTerminal } from '#shared/jobs'
const rows = ref(20_000)
const { job, error, connected, start } = useJob()
const busy = computed(() => !!job.value && !isTerminal(job.value))
const percent = computed(() => Math.round((job.value?.progress ?? 0) * 100))
</script>
<template>
<main class="export">
<label>
Rows
<input v-model.number="rows" type="number" min="1" max="200000">
</label>
<button :disabled="busy" @click="start(rows)">
{{ busy ? 'Exporting…' : 'Export' }}
</button>
<section v-if="job" class="status">
<progress :value="job.progress" max="1" />
<p>{{ percent }}% — {{ job.message }}</p>
<p v-if="job.status === 'done' && job.result">
<a :href="job.result.url">Download ({{ Math.round(job.result.bytes / 1024) }} KB)</a>
</p>
<p v-else-if="job.status === 'failed'" class="bad">
{{ job.error }}
</p>
<p v-else-if="!connected" class="muted">
Reconnecting…
</p>
</section>
<p v-if="error" class="bad">{{ error }}</p>
</main>
</template>
<style scoped>
.export { display: grid; gap: 1rem; max-width: 32rem; padding: 2rem; }
progress { width: 100%; height: .75rem; }
.bad { color: #c0392b; }
.muted { opacity: .6; }
</style>
Note the shape of the failure states. job.error is a message the task chose to expose; error is a transport problem the composable noticed. They're different things and users read them differently — one means "your export broke", the other means "we lost the narration, the export is probably still going." Conflating them produces a UI that panics about a job that's fine.
And note what the component doesn't do: it has no polling, no setInterval, no retry logic and no knowledge of Redis. It renders a Job.
10) The reaper
Jobs get stuck. The process is SIGKILLed mid-drain, a job is left running at 40%, and nothing will ever move it — the queue array has already forgotten it, and the drain only ever looks forward.
// server/tasks/jobs/reap.ts
import type { Job } from '#shared/jobs'
const STUCK_AFTER = 15 * 60 * 1000
const KEEP_FINISHED = 24 * 60 * 60 * 1000
export default defineTask({
meta: { name: 'jobs:reap', description: 'Fail stuck jobs and delete old ones' },
async run() {
const store = useStorage<Job>('jobs')
// Only the job records — not the `queue` key living alongside them.
const keys = await store.getKeys('job')
const now = Date.now()
let failed = 0
let deleted = 0
for (const key of keys) {
const job = await store.getItem(key)
if (!job) continue
if (job.status === 'running' && now - (job.startedAt ?? job.createdAt) > STUCK_AFTER) {
await patchJob(job.id, {
status: 'failed',
message: 'Abandoned',
error: 'The job stopped responding and was cancelled',
finishedAt: now,
})
failed++
}
else if (job.finishedAt && now - job.finishedAt > KEEP_FINISHED) {
await store.removeItem(key)
deleted++
}
}
return { result: { failed, deleted, scanned: keys.length } }
},
})
getKeys('job') prefixes the scan, which on the Redis driver becomes a SCAN with a match pattern rather than a KEYS *. That distinction matters at scale — KEYS blocks the whole Redis server while it walks the keyspace. On the fs driver it's a directory listing. Either way, this is an hourly task and not a hot path.
You can run it by hand while the dev server is up, which is the fastest way to test it:
npx nitro task run jobs:reap
# or hit the dev-only endpoint: GET /_nitro/tasks/jobs:reap
/_nitro/tasks lists everything registered, including the cron schedule — a good first check when a task "isn't running" and you're not sure whether it was even found.
11) What breaks in production
This is the part that turns a working demo into a working deployment.
Nginx. h3 sends X-Accel-Buffering: no, so buffering is already handled — most SSE-behind-Nginx advice on the internet is solving a problem you don't have. What you do still need is HTTP/1.1 with a cleared Connection header, and a read timeout longer than your longest job:
location /api/ {
proxy_pass http://127.0.0.1:3000;
proxy_http_version 1.1;
proxy_set_header Connection '';
proxy_read_timeout 1h;
}
The default proxy_read_timeout is 60 seconds, and it counts silence between reads. A job that goes two minutes without changing version sends nothing, and Nginx closes an idle connection that was perfectly healthy. Either raise the timeout or send a heartbeat every thirty seconds — three lines in tick(), and the more robust of the two. The classic SSE heartbeat is a bare comment line, but h3's push() always emits a data: field, so push(':keepalive') produces data: :keepalive rather than a comment. Use a named no-op instead — stream.push({ event: 'ping', data: '' }) — and ignore it on the client. If you're deploying this way, the PM2 and Nginx setup is where these blocks go.
Multiple instances. PM2 cluster mode with four workers gives you four croners, four jobs:drain schedules, and four in-flight-task maps that know nothing about each other. Every minute, four drains race for the same queue. Fix it at whichever layer you prefer: run the scheduler on one instance only (pm2 start --instances 1 for a dedicated worker process, app servers scaled separately), or make claimNext atomic with the LPOP from section 3. Doing both is not overkill.
Serverless. On Cloudflare the cron fires a fresh invocation with its own limits, so BUDGET_MS has to fit inside them: a paid Cron Trigger gets 30 seconds of CPU (10ms on the free plan) against a 15-minute wall clock, so 50 seconds of drain is fine only if most of it is spent waiting on I/O rather than computing. Vercel needs the cron declared in vercel.json and an endpoint that calls runTask itself, because Nitro 2.12 doesn't wire scheduledTasks up for that preset at all. And SSE holds a connection open for the entire job, which on per-invocation pricing you are paying for by the second. Long-lived streams and serverless are an awkward pairing; a Node process on a small VM is the natural home for this design.
Browser connection limits. HTTP/1.1 allows six connections per origin, and an open EventSource occupies one for its lifetime. Three tabs watching three jobs is fine; a dashboard opening one stream per row in a table will deadlock the origin — including your normal $fetch calls, which is a spectacularly confusing bug. HTTP/2 raises the limit to around a hundred and is what you're on in production behind TLS. The design fix is one stream carrying all the jobs a page cares about rather than one stream per job.
The experimental flag. Nitro tasks have carried it for a couple of years and the API has been stable throughout, but Nitro 3 moves the imports around (nitro/task, nitro/storage) as part of the general h3 v2 migration. Keep runTask behind your own thin wrapper if you want that to be a one-file change.
Where to take it
- Retries with backoff.
attemptsis already on theJoband unused. On failure, ifattempts < 3, push the id back onto the queue with anotBeforetimestamp instead of marking it failed, and haveclaimNextskip jobs that aren't due yet. - Cancellation. Add a
cancelRequestedflag, check it at the top of each chunk inbuildReport, and you get a Cancel button — the chunked loop from section 4 is already the right shape for it. - Redis pub/sub instead of polling.
redisClient()gets you an ioredis instance;duplicate()it for the subscriber, publish onjob:<id>frompatchJob, and the 250ms tick becomes an event. Worth it above roughly a thousand concurrent streams. - One stream for many jobs. Instead of
/api/jobs/:id/events, a/api/jobs/eventsthat streams every job belonging to the current user. Solves the connection-limit problem and is less code than it sounds. - Real output. Swap
buildReportfor a Drizzle query streaming into a CSV — the type-safe data layer build is the other end of exactly this pipeline. - BullMQ, when you outgrow this. Priorities, repeatable jobs, flows, a real dashboard. The
Jobinterface and the SSE half survive the migration; onlyserver/utils/queue.tschanges.
Wrapping up
Five files: a type in shared/, a queue over useStorage, a task that drains it, a handler that streams progress, and a composable that listens. No broker, no worker container, no job library.
Three things to carry into your own build. Put a monotonic version on anything you're going to stream, because it makes change detection an integer comparison and gives you a correct SSE event id for free. Understand that runTask deduplicates by name — it's what makes a single-process queue safe, and it's why you name tasks after workers rather than after jobs. And always send an explicit end event, because EventSource will otherwise reconnect to a finished job until the user closes the tab.
The larger point is that "we need a job queue" usually gets answered with infrastructure, and the answer is often a design change instead. The request stopped doing the work. Everything else followed from that.
