Real-Time Job Progress with WebSockets
In this tutorial, you will learn about Real. We cover key concepts, practical examples, and best practices to help you master this topic.
Track background job progress in real-time using WebSockets and SSE, push status updates to clients, and display percentage completion with live dashboards.
What You Learn
You will learn how to implement progress tracking with Redis pub/sub, push updates via Websocket and SSE, build a progress dashboard, and handle reconnection after client disconnect.
Why It Matters
Users want to know what is happening. A job that takes 30 seconds with no feedback feels broken. Progress tracking improves user experience and provides transparency into long-running operations.
Real-World Use
DodaTech's file upload scanner shows a progress bar: "Scanning file (45%) - Checking signatures..." The scanner publishes progress to Redis pub/sub, and the web UI receives updates via WebSocket.
Progress Tracking Architecture
flowchart LR
W[Worker] -->|Publish progress| RP[Redis Pub/Sub]
RP -->|Channel: job:progress| WS[WebSocket Server]
WS -->|Push update| C1[Client 1]
WS -->|Push update| C2[Client 2]
C1 -->|Subscribe job_id| WS
C2 -->|Subscribe job_id| WS
W -.->|Direct SSE| SSE[SSE Endpoint]
SSE --> C3[Client 3]
Redis Pub/Sub Progress Tracking
import redis
import json
import time
import threading
r = redis.Redis()
class ProgressTracker:
def __init__(self):
self.pubsub = r.pubsub()
def create_job(self, job_id, total_steps):
job_key = f'job:progress:{job_id}'
r.hset(job_key, mapping={
'status': 'running',
'current_step': 0,
'total_steps': total_steps,
'percent': 0,
'message': 'Starting...',
})
r.expire(job_key, 3600)
return job_id
def update_progress(self, job_id, current_step, total_steps, message):
percent = int((current_step / total_steps) * 100)
job_key = f'job:progress:{job_id}'
r.hset(job_key, mapping={
'current_step': current_step,
'total_steps': total_steps,
'percent': percent,
'message': message,
'updated_at': time.time(),
})
r.publish(f'job:updates:{job_id}', json.dumps({
'job_id': job_id,
'percent': percent,
'current_step': current_step,
'total_steps': total_steps,
'message': message,
}))
return percent
def complete_job(self, job_id):
r.hset(f'job:progress:{job_id}', 'status', 'completed')
r.publish(f'job:updates:{job_id}', json.dumps({
'job_id': job_id,
'status': 'completed',
'percent': 100,
}))
def fail_job(self, job_id, error):
r.hset(f'job:progress:{job_id}', 'status', 'failed')
r.publish(f'job:updates:{job_id}', json.dumps({
'job_id': job_id,
'status': 'failed',
'error': error,
}))
def get_progress(self, job_id):
data = r.hgetall(f'job:progress:{job_id}')
if data:
return {k.decode(): v.decode() if isinstance(v, bytes) else v
for k, v in data.items()}
return None
tracker = ProgressTracker()
job_id = tracker.create_job('scan-001', 10)
def simulate_work():
for i in range(1, 11):
time.sleep(0.2)
tracker.update_progress(job_id, i, 10, f"Scanning chunk {i}/10")
tracker.complete_job(job_id)
threading.Thread(target=simulate_work, daemon=True).start()
time.sleep(0.5)
progress = tracker.get_progress(job_id)
print(f"Progress: {progress['percent']}% - {progress['message']}")
time.sleep(2)
progress = tracker.get_progress(job_id)
print(f"Progress: {progress['percent']}% - {progress['message']}")
Expected output:
Progress: 20% - Scanning chunk 2/10
Progress: 70% - Scanning chunk 7/10
WebSocket Progress Server
import asyncio
import json
import redis.asyncio as aioredis
import time
async def websocket_progress_handler(websocket, job_id):
r = aioredis.Redis()
pubsub = r.pubsub()
await pubsub.subscribe(f'job:updates:{job_id}')
try:
async for message in pubsub.listen():
if message['type'] == 'message':
data = json.loads(message['data'])
await websocket.send(json.dumps(data))
if data.get('status') in ('completed', 'failed'):
break
except Exception as e:
print(f"WebSocket error: {e}")
finally:
await pubsub.unsubscribe(f'job:updates:{job_id}')
await r.close()
# Example usage
# Client connects to ws://server/ws/job/scan-001
# Server calls websocket_progress_handler(websocket, 'scan-001')
Expected output:
{"job_id": "scan-001", "percent": 10, "message": "Scanning chunk 1/10"}
{"job_id": "scan-001", "percent": 20, "message": "Scanning chunk 2/10"}
...
Server-Sent Events (SSE) Progress
import json
import time
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
class SSEProgressHandler(BaseHTTPRequestHandler):
job_id = None
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self.end_headers()
r = redis.Redis()
pubsub = r.pubsub()
pubsub.subscribe(f'job:updates:{self.job_id}')
try:
for message in pubsub.listen():
if message['type'] == 'message':
data = message['data'].decode()
self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush()
if '"completed"' in data or '"failed"' in data:
break
except Exception:
pass
finally:
pubsub.unsubscribe()
# Example: http://server/events/job/scan-001
class ProgressEventSource:
def __init__(self):
self.handlers = {}
def send_event(self, job_id, data):
event_key = f'sse:{job_id}'
r = redis.Redis()
r.publish(f'job:updates:{job_id}', json.dumps(data))
def listen(self, job_id):
r = redis.Redis()
pubsub = r.pubsub()
pubsub.subscribe(f'job:updates:{job_id}')
events = []
def collector():
for msg in pubsub.listen():
if msg['type'] == 'message':
events.append(json.loads(msg['data']))
if msg['data'].get('status') in ('completed', 'failed'):
break
t = threading.Thread(target=collector, daemon=True)
t.start()
return events
prog = ProgressEventSource()
def run_job():
for i in range(1, 6):
time.sleep(0.2)
prog.send_event('scan-002', {'percent': i*20, 'step': i})
prog.send_event('scan-002', {'status': 'completed'})
threading.Thread(target=run_job, daemon=True).start()
time.sleep(0.3)
print("SSE events sent to channel job:updates:scan-002")
Expected output:
SSE events sent to channel job:updates:scan-002
Progress Dashboard Provider
import redis
import json
import time
r = redis.Redis()
class ProgressDashboard:
def __init__(self):
self.active_jobs_key = 'progress:active_jobs'
def register_job(self, job_id, name, total_steps):
r.sadd(self.active_jobs_key, job_id)
r.hset(f'progress:meta:{job_id}', mapping={
'name': name,
'total_steps': total_steps,
'started_at': time.time(),
})
def update(self, job_id, current_step, message):
percent = int((current_step / int(r.hget(f'progress:meta:{job_id}', 'total_steps'))) * 100)
r.hset(f'progress:status:{job_id}', mapping={
'percent': percent,
'message': message,
'current_step': current_step,
'updated_at': time.time(),
})
r.publish('progress:channel', json.dumps({
'job_id': job_id,
'percent': percent,
'message': message,
}))
def complete(self, job_id):
r.hset(f'progress:status:{job_id}', 'percent', 100)
r.hset(f'progress:status:{job_id}', 'message', 'Completed')
r.srem(self.active_jobs_key, job_id)
def get_all_active(self):
job_ids = r.smembers(self.active_jobs_key)
jobs = []
for jid in job_ids:
jid = jid.decode()
meta = r.hgetall(f'progress:meta:{jid}')
status = r.hgetall(f'progress:status:{jid}')
jobs.append({
'id': jid,
'name': meta.get(b'name', b'').decode(),
'percent': int(status.get(b'percent', 0)),
'message': status.get(b'message', b'').decode(),
})
return jobs
def get_job_detail(self, job_id):
meta = r.hgetall(f'progress:meta:{job_id}')
status = r.hgetall(f'progress:status:{job_id}')
if not meta:
return None
return {
'id': job_id,
'name': meta.get(b'name', b'').decode(),
'total_steps': int(meta.get(b'total_steps', 0)),
'percent': int(status.get(b'percent', 0)),
'message': status.get(b'message', b'').decode(),
'started_at': float(meta.get(b'started_at', 0)),
}
dash = ProgressDashboard()
dash.register_job('scan-003', 'Security Scan', 20)
dash.update('scan-003', 5, 'Scanning system files...')
print(dash.get_all_active())
detail = dash.get_job_detail('scan-003')
print(f"Job: {detail['name']} - {detail['percent']}%")
Expected output:
[{'id': 'scan-003', 'name': 'Security Scan', 'percent': 25, 'message': 'Scanning system files...'}]
Job: Security Scan - 25%
Common Mistakes
1. Polling Instead of Pushing
Clients polling every second wastes bandwidth and creates latency. Use WebSocket or SSE to push updates only when progress changes.
2. No Reconnection Logic
If a client disconnects and reconnects, they miss progress updates. Send the latest state on reconnection before subscribing to new updates.
3. Too Frequent Updates
Sending progress on every iteration floods the network. Batch updates or throttle to at most once per 100ms or per percentage point.
4. Blocking the Worker with Progress IO
Workers slow down when publishing progress synchronously. Use async pub/sub or fire-and-forget progress updates to avoid blocking job execution.
5. No Cleanup for Completed Jobs
Progress keys accumulate in Redis. Set TTL or clean up after a grace period to prevent memory leaks.
Practice Questions
1. What is the difference between WebSocket and SSE for progress tracking?
WebSocket is bidirectional and better for real-time dashboards. SSE is simpler, unidirectional from server to client, and works over standard HTTP.
2. How does Redis pub/sub enable real-time progress?
Workers publish progress messages to Redis channels. WebSocket servers subscribe to those channels and forward messages to connected clients.
3. Why should progress updates be throttled?
Sending updates on every iteration creates high message volume. Throttling to 100ms intervals or 1% changes reduces load while maintaining smooth UX.
4. How do you handle client reconnection?
On reconnect, send the current job state before subscribing to new updates. The client can then resume receiving live progress.
Challenge
Build a real-time progress system for a batch file processor: worker publishes progress every chunk, WebSocket server pushes to subscribed clients, client shows progress bar with percentage and message, handles reconnection, and shows history for completed jobs.
FAQ
Mini Project: Real-Time Progress System
import redis
import json
import time
import threading
r = redis.Redis()
class RealTimeProgress:
def __init__(self):
self.channels = {}
def start_job(self, job_id, total, name):
r.hset(f'rtp:{job_id}', mapping={
'name': name, 'total': total, 'current': 0,
'percent': 0, 'status': 'running', 'started': time.time()
})
def update(self, job_id, current, message):
total = int(r.hget(f'rtp:{job_id}', 'total'))
percent = int((current / total) * 100)
r.hset(f'rtp:{job_id}', mapping={
'current': current, 'percent': percent, 'message': message
})
r.publish(f'rtp:channel:{job_id}', json.dumps({
'job_id': job_id, 'percent': percent,
'current': current, 'total': total, 'message': message
}))
def finish(self, job_id, status='completed'):
r.hset(f'rtp:{job_id}', 'status', status)
r.publish(f'rtp:channel:{job_id}', json.dumps({'status': status, 'job_id': job_id}))
def get_state(self, job_id):
data = r.hgetall(f'rtp:{job_id}')
if not data:
return None
return {k.decode(): v.decode() for k, v in data.items()}
rtp = RealTimeProgress()
rtp.start_job('batch-1', 50, 'File Scan')
def worker_sim():
for i in range(1, 51):
time.sleep(0.05)
rtp.update('batch-1', i, f"Processing file {i}/50")
rtp.finish('batch-1')
threading.Thread(target=worker_sim, daemon=True).start()
time.sleep(0.3)
state = rtp.get_state('batch-1')
print(f"Progress: {state['percent']}% - {state['message']}")
time.sleep(1)
state = rtp.get_state('batch-1')
print(f"Progress: {state['percent']}% - {state['message']}")
Expected output:
Progress: 12% - Processing file 6/50
Progress: 52% - Processing file 26/50
What's Next
Now that you understand progress tracking, explore job timeout handling for preventing stuck jobs, then learn about job lifecycle events for comprehensive job monitoring.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro