Progress & Streaming¶

TaskQ provides a real-time progress pipeline: actors emit structured updates via ctx.progress(), a worker-side buffer coalesces and flushes state to Postgres, and a Redis pub/sub bridge delivers events to subscribers immediately. Consumers can receive events server-to-client via:

JobHandle.progress_stream(): Python async iterator (worker or service process)
HTTP SSE endpoint: browser or HTTP client via GET /api/job/{job_id}/progress/stream relative to wherever the router is mounted. Under taskq ui serve (the admin UI), the admin router mounts this progress router at /jobs, and the admin router itself mounts at /admin, so the effective path is GET /admin/jobs/api/job/{job_id}/progress/stream (and /admin/jobs/api/job/{job_id}/state). See Mounting the router.

Redis is optional

When the [redis] extra is not installed or TASKQ_REDIS_URL is not set:

ctx.progress() still works: updates are coalesced and flushed to Postgres.
JobHandle.progress_stream() falls back to 500 ms Postgres polling (higher latency, same data).
The HTTP SSE endpoint returns HTTP 503 with {"error": "redis_not_configured"}.
TaskQ.stream() falls back to 500 ms Postgres polling of the job row (a transition is seen within half a second, no Redis required).

Install Redis for immediate event delivery:

pip install "taskq-py[redis]"

How progress works¶

The actor calls await ctx.progress(step=…, percent=…).
The per-job in-memory buffer is updated synchronously (last-writer-wins per field) and seq is incremented.
A kind="progress" event is scheduled as a background task to publish to the Redis channel {schema}:progress:{job_id}. ctx.progress() returns as soon as the buffer is updated; does not wait for the publish to complete or even start.
A periodic flush loop writes the latest coalesced state to the jobs.progress_state JSONB column and jobs.progress_seq counter.
At job completion or crash, the buffer is flushed one final time before the terminal status is written.

The seq is the event stream's total order. Every event on a job's stream carries a seq strictly greater than every event before it, progress events and state-change events alike: state-change events (the running transition at dispatch, retries, snoozes, and the terminal exits succeeded / failed / cancelled / abandoned / interrupted releases) each consume the next seq value rather than repeating the last progress event's. A consumer deduping or resuming by seq alone (the Last-Event-ID discipline below) therefore sees every event exactly once, and the terminal event always arrives strictly after the last progress event it supersedes. The durable jobs.progress_seq carries the same total order: it ends at the terminal event's consumed seq. The snapshot guarantee is scoped to events whose durable write has landed: a reconnecting consumer resuming from the Postgres snapshot never rewinds past any event whose write reached jobs.progress_seq. One window to know about: the running transition consumes its seq in memory (and publishes it to Redis) but is deliberately not durable on its own, so a job whose actor never calls ctx.progress() holds progress_seq == 0 until its terminal write, and a client that saw running and reconnects before the next flush or the terminal finds the snapshot unchanged. The live stream is unaffected; only the Postgres snapshot lags until the next durable write.

The total order holds within a worker's lifetime, not across worker death. The reclaim sweep does not touch jobs.progress_seq, so if a worker dies mid-job the reclaimed job's next attempt re-publishes running at a seq the wire has already carried (a terminal publish whose mark_* write failed does the same when the job retries). Across worker death a seq-cursor consumer can therefore see the same seq twice with different payloads; deduping by seq is safe for events whose durable write landed, and the Postgres snapshot remains authoritative across the gap. During a rolling deploy, workers still on the previous release emit state-change events that repeat the head instead of consuming a new seq until they are upgraded; a seq-cursor guard drops those as duplicates, which is the old, safe behaviour.

Redis publishes are coalesced: at most one publish per job is in flight at a time; a call racing a running publish is latched onto the buffer and the running task re-publishes the latch when its round trip lands, so the final publish always lands. Postgres retains only the most recent snapshot, intermediate events are replaceable by a later seq, and clients reconnecting via Last-Event-ID receive a catch-up snapshot from Postgres and then resume the live Redis stream.

The Redis publish is fire-and-forget. Because it runs as a background task, ctx.progress() never blocks the calling actor code on the network; this holds even when an actor calls ctx.progress() at high frequency in a tight loop. It is safe by design for the publish to complete out of order relative to other in-flight publishes for the same job, or to be dropped outright (e.g. on a transient Redis error): consumers of the SSE/pub-sub stream already discard any event whose seq is not strictly greater than the last one they've seen, so an out-of-order or missing event never corrupts displayed state. The Postgres-persisted progress_state / progress_seq (flushed on the periodic coalesce interval described above) remains the durable source of truth regardless of what happens on the Redis side. Failures publishing to Redis are logged and recorded as a metric, never raised to the caller.

Reporting progress from an actor¶

from taskq import actor
from taskq.context import JobContext


@actor(queue="media")
async def transcode_video(payload: TranscodePayload, ctx: JobContext[TranscodePayload]) -> None:
    segments = await split_into_segments(payload.url)
    total = len(segments)

    for i, segment in enumerate(segments):
        await transcode_segment(segment)
        await ctx.progress(
            step=i + 1,
            percent=round((i + 1) / total * 100, 1),
            detail=f"Segment {i + 1}/{total}",
        )

    await ctx.progress(percent=100.0, detail="Done")

See ctx.progress() in the Actor API for the full parameter reference.

Consuming progress in Python¶

Use JobHandle.progress_stream() to iterate events in a service or worker process:

handle = await client.enqueue(transcode_video, payload)

async for event in handle.progress_stream():
    if event.percent is not None:
        print(f"  {event.percent:.0f}%: {event.detail}")
    if event.terminal:
        print(f"finished: {event.status}")
        break

progress_stream() yields ProgressEvent objects and stops automatically when a terminal=True event is received. The first event is the row as it stands when the stream opens — the same contract TaskQ.stream() documents — so a consumer attaching to an already-running job does not starve until the actor's next publish.

Failure budget. A blip shorter than the transport's failure budget is retried silently (beyond a warning per failed poll) — on the Postgres poll transport that is every fetch, and on the Redis path, the safety-net re-reads behind the subscription. If every fetch fails for at least the budget, the stream ends by raising StreamUnavailable — the caller's async for cannot wait forever on a database that is not coming back; the last failure is chained as __cause__. See the Exceptions API reference.

`ProgressEvent` fields¶

Field	Type	Description
`kind`	`"progress" \\| "state_change"`	Event class: `"progress"` is an incremental progress update, `"state_change"` a status transition or terminal exit.
`job_id`	`UUID`	Job this event belongs to.
`actor`	`str`	Actor name.
`ts`	`datetime`	Server-side timestamp.
`seq`	`int`	Strictly-monotone sequence number: the total order over the job's whole event stream, state-change events included.
`status`	`str`	Current job status at publish time.
`step`	`int \\| None`	Step counter.
`percent`	`float \\| None`	Completion percentage.
`detail`	`str \\| None`	Human-readable status string.
`data`	`dict[str, object] \\| None`	Custom structured data.
`terminal`	`bool`	`True` when job has reached a terminal state.

Transport. When Redis is configured, events are delivered via Redis pub/sub with no polling delay. Without Redis the method falls back to polling Postgres at 500 ms intervals. NotImplementedError is raised against InMemoryBackend.

HTTP SSE endpoint¶

The taskq.web.progress module provides a FastAPI router with two routes:

Method	Path	Description
`GET`	`/api/job/{job_id}/progress/stream`	Server-Sent Events stream
`GET`	`/api/job/{job_id}/state`	One-shot JSON poll

SSE stream¶

GET /api/job/{job_id}/progress/stream

Query parameters:

Parameter	Type	Description
`last_event_id`	`int \\| None`	Resume from this sequence number. Also read from the `Last-Event-ID` request header (WHATWG EventSource spec). Both sources get the same validation (see the `400` row below).

SSE event types:

`event`	`id` present	Description
`progress`	yes (`seq`)	Incremental progress update. `data` is a JSON-serialised `ProgressEvent`.
`terminal`	yes (`seq`)	Job reached a terminal state. `data` is a JSON-serialised `ProgressEvent`. Close the connection after receiving this.
`done`	no	Stream is closing. Close the connection.
`: keepalive`	n/a	SSE comment emitted every 15 s (configurable). No `event` field.

HTTP status codes:

Code	Meaning
`200`	Stream established.
`400`	The resume cursor — the `Last-Event-ID` header or the `?last_event_id=` query parameter — is not a non-negative integer no greater than 2^31-1. The `400` names the rejected source.
`404`	Job not found.
`503`	Redis not configured or unavailable. `Retry-After: 2` header is set.

Reconnect semantics. The browser's native EventSource sends Last-Event-ID on reconnect automatically. The endpoint subscribes to Redis before querying Postgres so there is no race window: if an event arrived between the disconnect and the reconnect it is caught by the Redis subscription. A catch-up snapshot is emitted from Postgres when progress_seq > last_event_id. The header wins over ?last_event_id= when both are present: the header is validated first and, if valid, the query parameter is never read. Both sources get the same validation — a non-negative integer no greater than 2^31-1 — and a violation is rejected with 400 naming the source, rather than read as "no cursor". The ceiling is wire hygiene bounding the domain a hostile cursor can name, not the storage domain: the durable progress_seq column is a bigint, and every id this stream issues sits far below it. (Before this contract reached the query parameter, a negative ?last_event_id= was silently clamped to zero and answered with a full catch-up snapshot; it now gets the header's 400.) Because state-change events consume seq, a consumer that reconnects after the last progress event still finds the row's progress_seq one ahead (the terminal consumed it) and receives the terminal snapshot instead of hanging on a stream nothing further is published to.

Mounting the router¶

from taskq.web.progress import create_router

progress_router = create_router(
    pg_pool,  # asyncpg.Pool
    redis_client,  # redis.asyncio.Redis | None
    schema="taskq",  # must match PostgresBackend schema
    auth_dependency=require_authenticated_user,  # required outside dev (fails closed)
    sse_heartbeat_interval=timedelta(seconds=15),
)

app.include_router(progress_router, prefix="/jobs")
# Produces:
#   GET /jobs/api/job/{job_id}/progress/stream
#   GET /jobs/api/job/{job_id}/state

Fail-closed outside dev. The auth_dependency=None default is not usable in a non-dev environment: create_router() raises RuntimeError when TASKQ_ENVIRONMENT is anything other than dev or development (the same fail-closed gate as the admin UI; see admin-ui.md). Opt out only when an authenticating ingress covers the mount point:

export TASKQ_PROGRESS_REQUIRE_AUTH=false
# WARNING log: progress-router-no-auth: but the router is created

Serving without auth in any environment (dev included) logs the progress-router-no-auth warning: each anonymous stream holds a Redis pubsub subscription and an asyncio task for as long as the client stays connected.

Under taskq ui serve / the admin UI. create_router() in taskq.web.admin mounts this same progress router internally at /jobs (src/taskq/web/admin/_factory.py), and the admin router is itself mounted at /admin (taskq ui serve / docs/guides/admin-ui.md). The resulting paths are GET /admin/jobs/api/job/{job_id}/progress/stream and GET /admin/jobs/api/job/{job_id}/state, not the bare /api/job/... or /jobs/api/job/... paths shown above, which only apply when you mount taskq.web.progress.create_router() yourself at a different prefix. Without Redis configured, the stream endpoint returns 503 {"error": "redis_not_configured"} while the /state poll endpoint still works.

`create_router()` parameters¶

Parameter	Type	Default	Description
`pg_pool`	`asyncpg.Pool`	required	Connection pool for snapshot reads.
`redis_client`	`redis.asyncio.Redis \\| None`	required	Redis client. Pass `None` to disable streaming (SSE returns 503).
`schema`	`str`	`"taskq"`	PostgreSQL schema; must match the backend.
`auth_dependency`	`Callable \\| None`	`None`	FastAPI `Depends`-compatible callable applied to all routes. Outside dev (`TASKQ_ENVIRONMENT` not `dev`/`development`) the factory raises `RuntimeError` when omitted: the default is not usable in non-dev unless `TASKQ_PROGRESS_REQUIRE_AUTH=false` suppresses the check (only for mounts behind an authenticating ingress). Serving without auth always logs a `progress-router-no-auth` warning.
`session_verifier`	`Callable[[Request], Awaitable[bool]] \\| None`	`None`	Async re-check for long-lived streams (#316): invoked before every streamed event and at every keepalive tick, a `False` (or raising) result ends the stream, so a session revoked mid-stream stops receiving frames within one keepalive interval, capped at 60 s. When omitted, the router derives it from the `session_verifier` attribute taskq's `create_auth_dependency`/`token_auth` attach to the callable they return; a custom `auth_dependency` without that attribute logs a `progress-stream-no-session-verifier` warning and streams authenticate once, at subscribe only.
`sse_heartbeat_interval`	`timedelta`	`timedelta(seconds=15)`	Interval for keepalive SSE comments. The keepalive tick is also the session re-check cadence, so the effective interval is capped at 60 s: a revoked session ends a quiet stream within at most 60 s, whatever this is set to.
`max_sse_connections`	`int \\| None`	`None` (`TASKQ_PROGRESS_MAX_SSE_CONNECTIONS`, default 50)	Maximum concurrent progress streams this process will serve; further connections get HTTP 429. Each stream holds a Redis pubsub subscription and an asyncio task for as long as the client stays connected. See admin-ui.md: SSE connection limits.
`resolve_pg_pool`	`Callable[[Request], asyncpg.Pool] \\| None`	`None`	Per-request resolver for the pool, used instead of `pg_pool` when given. A host whose pool is replaced while it runs (`taskq ui serve` rebuilds it on a credential rotation) resolves the live one per request. A FastAPI dependency: it may declare `request: Request`.
`resolve_redis_client`	`Callable[[Request], Any] \\| None`	`None`	Per-request resolver for the Redis client, used instead of `redis_client` when given — same credential-rotation pattern as `resolve_pg_pool`. A FastAPI dependency: it may declare `request: Request`.

Browser client example¶

const jobId = "018f1a2b-3c4d-7e5f-8a9b-0c1d2e3f4a5b";
const url = `/jobs/api/job/${jobId}/progress/stream`;

const es = new EventSource(url);

es.addEventListener("progress", (e) => {
  const event = JSON.parse(e.data);
  document.getElementById("progress").textContent =
    `${event.percent ?? "?"}%: ${event.detail ?? ""}`;
});

es.addEventListener("terminal", (e) => {
  const event = JSON.parse(e.data);
  console.log("job finished:", event.status);
  es.close();
});

es.addEventListener("done", () => es.close());

es.onerror = (err) => {
  // EventSource reconnects automatically on transient errors.
  // The server uses Last-Event-ID to resume from where streaming left off.
  console.warn("SSE connection error, reconnecting…", err);
};

The browser's EventSource API handles reconnection automatically and sends Last-Event-ID on each reconnect so no progress events are lost.

Poll-state fallback¶

For environments where SSE is not available (e.g. HTTP/1.1 proxies that buffer responses, or clients that do not support EventSource), the poll-state endpoint returns the current snapshot as JSON:

GET /api/job/{job_id}/state

Response body:

{
  "status": "running",
  "progress_state": {
    "step": 3,
    "percent": 60.0,
    "detail": "Segment 3/5"
  },
  "progress_seq": 3
}

Poll this endpoint at whatever interval suits your UI. Use progress_seq to detect staleness between polls.

Gotchas & best practices¶

The Redis publish is fire-and-forget

ctx.progress() never blocks on the network, even under a tight loop calling it many times per second. The trade-off is that individual Redis publishes may arrive out of order or be dropped. This is safe: SSE/pub-sub consumers discard any event whose seq does not exceed the last one seen, and the periodically-flushed Postgres progress_state/progress_seq is always the durable source of truth: a dropped Redis publish never loses progress data, it only delays a subscriber's view of it until the next successful publish or the next poll.

seq is a total order; state-change events consume it. Every state-change event (the dispatch running transition, retries, snoozes, and every terminal or interrupted exit) advances the same counter a progress event does, so deduping by seq alone is correct for events whose durable write landed. Two windows put duplicate seqs back on the wire: a worker dying mid-job re-publishes running (or a terminal) at a seq already carried, and a rolling deploy's not-yet-upgraded workers emit the old repeated-head state-change events until they are upgraded. Both produce duplicates a seq-cursor guard already drops; the Postgres snapshot stays authoritative across the gap. Consumers written against the older progress-only contract, where a state-change event repeated the last progress event's seq and a seq-cursor guard dropped it, need no change beyond removing any special-case handling that assumed the repeats: the extra, correctly-ordered seq values they now see are ordinary events.
Redis is optional, not required. Without TASKQ_REDIS_URL configured (or the [redis] extra installed), ctx.progress() still coalesces and flushes to Postgres exactly as described above; only the Redis publish step is skipped. JobHandle.progress_stream() transparently falls back to 500 ms Postgres polling, and the HTTP SSE endpoint returns 503 {"error": "redis_not_configured"} (the poll-state endpoint still works).
InMemoryBackend does not support streaming. progress_stream() raises NotImplementedError when the client is constructed against InMemoryBackend (used in unit tests). Use the Postgres-backed test fixtures if a test needs to exercise progress streaming end to end.
High-frequency progress in tight loops is fine. Because the Redis publish never blocks the actor and the coalesce buffer collapses intermediate values, calling ctx.progress() on every iteration of a hot loop is safe; only the periodic flush interval (TASKQ_PROGRESS_COALESCE_INTERVAL) bounds how much Postgres write traffic this generates, and each superseding call is latched so the latest state is what publishes.

Configuration¶

Environment variable	Default	Description
`TASKQ_REDIS_URL`	`None`	Redis connection URL. Required for real-time progress delivery.
`TASKQ_PROGRESS_DATA_MAX_BYTES`	`16384` (16 KiB)	Maximum serialised size of the `data` dict passed to `ctx.progress()`. Range: 1 KiB to 1 MiB. Exceeding this raises `ProgressTooLarge`. Snapshot-style publishers — actors that send the full document as `data` on every tick — must size the cap at or above their payload bound: `data` only grows, so once one snapshot exceeds the cap every later snapshot raises too, and the durable row freezes at the last accepted state until the job reaches a terminal status.
`TASKQ_PROGRESS_COALESCE_INTERVAL`	`0.5`	Seconds between periodic flush ticks that write coalesced state to Postgres. Minimum: 0.1 s.
`TASKQ_PROGRESS_PUBLISH_GLOBAL`	`true`	When `true`, every event is also published to a schema-wide fanout channel in addition to the per-job channel. Disable in high-throughput deployments without a global subscriber.

Install the redis extra to enable real-time delivery:

uv add "taskq-py[redis]"

Without the redis extra: - ctx.progress() still coalesces and flushes to Postgres but does not publish to Redis. - JobHandle.progress_stream() falls back to 500 ms Postgres polling. - The HTTP SSE endpoint returns HTTP 503.