# Job Queue System Low-Level Design: Worker Pool, Retry Backoff, Heartbeat, and Recurring Jobs Source: https://www.techinterview.org/post/3233468594/lld-job-queue/ Updated: 2026-04-18 · techinterview.org ## Job Queue System: Low-Level Design A job queue system schedules and executes background work: sending emails, resizing images, generating reports, syncing data with third-party APIs. Unlike a messaging queue (fire-and-forget delivery), a job queue tracks execution state — scheduled, running, completed, failed — and provides retry logic, job chaining, priority scheduling, and observability into worker throughput. This design covers the job lifecycle, worker pool management, job deduplication, and cron-style recurring jobs. ## Core Data Model ``` CREATE TABLE JobQueue ( queue_name VARCHAR(100) PRIMARY KEY, max_concurrency INT NOT NULL DEFAULT 10, max_retries SMALLINT NOT NULL DEFAULT 3, retry_backoff VARCHAR(20) NOT NULL DEFAULT 'exponential', -- exponential, linear, fixed created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE TABLE Job ( job_id BIGSERIAL PRIMARY KEY, queue_name VARCHAR(100) NOT NULL REFERENCES JobQueue(queue_name), job_type VARCHAR(100) NOT NULL, -- 'send_email', 'resize_image', 'sync_crm' payload JSONB NOT NULL, idempotency_key VARCHAR(200), -- deduplicate concurrent enqueues status VARCHAR(20) NOT NULL DEFAULT 'pending', -- pending, running, completed, failed, cancelled priority SMALLINT NOT NULL DEFAULT 0, attempt_count SMALLINT NOT NULL DEFAULT 0, run_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- earliest execution time started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, worker_id VARCHAR(200), result JSONB, last_error TEXT, parent_job_id BIGINT REFERENCES Job(job_id), -- for job chains created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ) PARTITION BY LIST (status); CREATE TABLE Job_pending PARTITION OF Job FOR VALUES IN ('pending'); CREATE TABLE Job_running PARTITION OF Job FOR VALUES IN ('running'); CREATE TABLE Job_completed PARTITION OF Job FOR VALUES IN ('completed'); CREATE TABLE Job_failed PARTITION OF Job FOR VALUES IN ('failed'); CREATE TABLE RecurringJob ( recurring_job_id SERIAL PRIMARY KEY, job_type VARCHAR(100) NOT NULL, queue_name VARCHAR(100) NOT NULL, payload JSONB NOT NULL DEFAULT '{}', cron_expression VARCHAR(100) NOT NULL, -- '0 9 * * 1-5' = weekdays 9am is_active BOOLEAN NOT NULL DEFAULT TRUE, last_run_at TIMESTAMPTZ, next_run_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE UNIQUE INDEX ON Job(queue_name, idempotency_key) WHERE idempotency_key IS NOT NULL AND status IN ('pending','running'); CREATE INDEX ON Job_pending(queue_name, priority DESC, run_at ASC); CREATE INDEX ON RecurringJob(next_run_at) WHERE is_active=TRUE; ``` ## Job Enqueue and Dedup ``` import json, datetime, socket def enqueue(job_type: str, payload: dict, queue_name: str = 'default', priority: int = 0, run_at: datetime.datetime = None, idempotency_key: str = None, parent_job_id: int = None) -> int: """ Enqueue a job. Returns job_id. Idempotent: if idempotency_key is provided and a job with that key is pending or running, returns the existing job_id. """ run_at = run_at or datetime.datetime.utcnow() try: row = db.fetchone(""" INSERT INTO Job (queue_name, job_type, payload, idempotency_key, priority, run_at, parent_job_id) VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (queue_name, idempotency_key) WHERE idempotency_key IS NOT NULL AND status IN ('pending','running') DO NOTHING RETURNING job_id """, (queue_name, job_type, json.dumps(payload), idempotency_key, priority, run_at, parent_job_id)) if not row: # Deduplicated — return the existing job_id existing = db.fetchone(""" SELECT job_id FROM Job WHERE queue_name=%s AND idempotency_key=%s AND status IN ('pending','running') """, (queue_name, idempotency_key)) return existing['job_id'] if existing else None return row['job_id'] except Exception as e: raise JobEnqueueError(str(e)) from e ``` ## Worker: Claim and Execute ``` import time, traceback, threading HEARTBEAT_INTERVAL = 10 # seconds; update run_at to prevent timeout def worker_loop(queue_name: str, handlers: dict, worker_id: str = None, batch_size: int = 5, poll_interval: float = 1.0): """ handlers: {'send_email': handle_send_email, 'resize_image': handle_resize_image} """ worker_id = worker_id or f"{socket.gethostname()}:{threading.get_ident()}" while True: jobs = _claim_jobs(queue_name, worker_id, batch_size) if not jobs: time.sleep(poll_interval) continue for job in jobs: _execute_job(job, handlers, worker_id) def _claim_jobs(queue_name: str, worker_id: str, batch_size: int) -> list: return db.fetchall(""" UPDATE Job SET status = 'running', started_at = NOW(), worker_id = %s, attempt_count = attempt_count + 1 WHERE job_id IN ( SELECT job_id FROM Job WHERE queue_name=%s AND status='pending' AND run_at <= NOW() ORDER BY priority DESC, run_at ASC LIMIT %s FOR UPDATE SKIP LOCKED ) RETURNING job_id, job_type, payload, attempt_count, parent_job_id """, (worker_id, queue_name, batch_size)) def _execute_job(job: dict, handlers: dict, worker_id: str): handler = handlers.get(job['job_type']) if not handler: _fail_job(job['job_id'], f"No handler for job_type: {job['job_type']}", job['attempt_count']) return # Heartbeat thread: keeps run_at fresh so job doesn't look timed out stop_heartbeat = threading.Event() heartbeat = threading.Thread( target=_heartbeat_loop, args=(job['job_id'], stop_heartbeat), daemon=True ) heartbeat.start() try: result = handler(job['payload']) stop_heartbeat.set() db.execute(""" UPDATE Job SET status='completed', completed_at=NOW(), result=%s WHERE job_id=%s """, (json.dumps(result) if result else None, job['job_id'])) # Enqueue chained jobs if this job is part of a chain _maybe_trigger_next(job['job_id'], job['parent_job_id']) except Exception as e: stop_heartbeat.set() _fail_job(job['job_id'], traceback.format_exc(), job['attempt_count']) def _heartbeat_loop(job_id: int, stop: threading.Event): while not stop.wait(HEARTBEAT_INTERVAL): db.execute( "UPDATE Job SET started_at=NOW() WHERE job_id=%s AND status='running'", (job_id,) ) def _fail_job(job_id: int, error: str, attempt_count: int): queue = db.fetchone(""" SELECT q.max_retries, q.retry_backoff FROM Job j JOIN JobQueue q USING(queue_name) WHERE j.job_id=%s """, (job_id,)) max_retries = queue['max_retries'] if queue else 3 if attempt_count < max_retries: # Retry with exponential backoff: 60s, 120s, 240s delay = 60 * (2 ** (attempt_count - 1)) next_run = datetime.datetime.utcnow() + datetime.timedelta(seconds=delay) db.execute(""" UPDATE Job SET status='pending', run_at=%s, last_error=%s WHERE job_id=%s """, (next_run, error[:1000], job_id)) else: db.execute(""" UPDATE Job SET status='failed', completed_at=NOW(), last_error=%s WHERE job_id=%s """, (error[:1000], job_id)) ``` ## Recurring Job Scheduler ``` from croniter import croniter def tick_recurring_jobs(): """ Run every minute (cron: * * * * *). Enqueues due recurring jobs and advances next_run_at. """ due = db.fetchall(""" SELECT * FROM RecurringJob WHERE is_active=TRUE AND next_run_at <= NOW() FOR UPDATE SKIP LOCKED """) for rj in due: # Enqueue with idempotency_key to prevent double-enqueue on concurrent ticks idem_key = f"recurring:{rj['recurring_job_id']}:{rj['next_run_at'].isoformat()}" enqueue( job_type=rj['job_type'], payload=rj['payload'], queue_name=rj['queue_name'], idempotency_key=idem_key, ) # Advance to next scheduled time cron = croniter(rj['cron_expression'], rj['next_run_at']) next_run = cron.get_next(datetime.datetime) db.execute(""" UPDATE RecurringJob SET last_run_at=%s, next_run_at=%s WHERE recurring_job_id=%s """, (rj['next_run_at'], next_run, rj['recurring_job_id'])) ``` ## Key Design Decisions - **Heartbeat prevents ghost jobs**: if a worker crashes mid-job, the job stays in status=running until a cleanup job detects it has timed out (started_at older than 2× heartbeat interval with no update). Without heartbeats, a crashed worker leaves a job in running state forever. The heartbeat pattern also supports very long-running jobs (report generation, ML training) without false-positive timeouts. - **Partition by status keeps active table small**: Job_pending and Job_running are queried on the hot path. Job_completed accumulates millions of rows. Partitioning lets you drop or archive Job_completed monthly without affecting the live queue index. - **Idempotency key dedup scope is pending+running only**: once a job completes, its idempotency key is no longer in scope — re-enqueuing the same idempotency key triggers a fresh run. This correctly models "run this job once per period" semantics without permanent dedup. - **Recurring job idempotency key includes next_run_at**: if the tick_recurring_jobs cron fires twice in the same minute (e.g., during a deploy restart), the same next_run_at value produces the same idempotency key — the second tick's enqueue is a no-op. Job queue and background processing system design is discussed in [Uber system design interview questions](https://www.techinterview.org/post/3233460273/uber-interview-guide-2026-dispatch-systems-geospatial-algorithms-and-marketplace-engineering/). Job queue and async task execution design is covered in [Airbnb system design interview preparation](https://www.techinterview.org/post/3233460279/airbnb-interview-guide-2026-search-systems-trust-and-safety-and-full-stack-engineering/). Job queue and distributed background job design is discussed in [Amazon system design interview guide](https://www.techinterview.org/post/3233460275/scale-ai-interview-guide-2026-data-infrastructure-rlhf-pipelines-and-ml-engineering/). See also: [Netflix Interview Guide 2026: Streaming Architecture, Recommendation Systems, and Engineering Excellence](https://www.techinterview.org/post/3233460274/netflix-interview-guide-2026-streaming-architecture-recommendation-systems-and-engineering-excellence/) See also: [Atlassian Interview Guide](https://www.techinterview.org/post/3233459919/atlassian-interview/)