Distributed Jobs: Run Tasks Across Servers

Problem

Modern applications need to process large volumes of tasks efficiently. Single-server job processing creates bottlenecks:

• Scalability Issues: A single server can only handle so many concurrent jobs
• Reliability Concerns: Server failure means job loss and service interruption
• Resource Constraints: CPU-intensive or long-running tasks block other operations
• Uneven Load Distribution: Some servers become overloaded while others sit idle
• Lack of Visibility: Difficult to monitor job status across infrastructure
• No Fault Tolerance: Failed jobs disappear without retry mechanisms

Consider an e-commerce platform processing 10,000 orders daily. Each order requires image processing, payment verification, and email notifications. A single server would struggle, causing delays and customer dissatisfaction.

Solution

Implement a distributed job queue system that:

1. Decouples Job Production from Execution: Producers enqueue jobs; workers process them independently
2. Horizontal Scaling: Add more workers to handle increased load
3. Load Balancing: Distribute jobs evenly across available workers
4. Fault Tolerance: Retry failed jobs automatically
5. Persistence: Store jobs in a message broker (Redis, RabbitMQ, AWS SQS)
6. Monitoring: Track job status, metrics, and worker health
7. Priority Queues: Process critical jobs first
8. Idempotency: Safely retry jobs without side effects

Architecture Components

┌─────────────────────────────────────────────────────┐
│                   Job Producers                      │
│  (Web Servers, APIs, Scheduled Tasks)               │
└────────────────┬────────────────────────────────────┘
                 │ Enqueue
                 ▼
┌─────────────────────────────────────────────────────┐
│            Message Broker / Queue                    │
│  (Redis, RabbitMQ, AWS SQS, Kafka)                  │
│  ┌──────────────┬──────────────┬──────────────┐    │
│  │ High Priority│ Normal Queue │ Low Priority │    │
│  └──────────────┴──────────────┴──────────────┘    │
└────────────────┬────────────────────────────────────┘
                 │ Dequeue
                 ▼
┌─────────────────────────────────────────────────────┐
│              Worker Pool (Scalable)                  │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ Worker 1 │ │ Worker 2 │ │ Worker N │            │
│  └──────────┘ └──────────┘ └──────────┘            │
└────────────────┬────────────────────────────────────┘
                 │ Results
                 ▼
┌─────────────────────────────────────────────────────┐
│         Result Storage / Callbacks                   │
│  (Database, Cache, Webhooks)                        │
└─────────────────────────────────────────────────────┘

Code Implementation

1. Job Queue Manager (Python with Redis)

import json
import uuid
import time
from datetime import datetime, timedelta
from typing import Any, Dict, Optional, List
from enum import Enum
import redis
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class JobStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    RETRYING = "retrying"

class JobPriority(Enum):
    LOW = 3
    NORMAL = 2
    HIGH = 1

class DistributedJobQueue:
    """Manages distributed job processing across multiple workers"""

    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            decode_responses=True
        )
        self.queue_prefix = "job:queue:"
        self.job_prefix = "job:data:"
        self.result_prefix = "job:result:"
        self.worker_prefix = "worker:"

    def enqueue_job(
        self,
        job_type: str,
        payload: Dict[str, Any],
        priority: JobPriority = JobPriority.NORMAL,
        max_retries: int = 3,
        timeout: int = 3600
    ) -> str:
        """Enqueue a new job"""
        job_id = str(uuid.uuid4())

        job_data = {
            "id": job_id,
            "type": job_type,
            "payload": json.dumps(payload),
            "status": JobStatus.PENDING.value,
            "priority": priority.value,
            "created_at": datetime.utcnow().isoformat(),
            "max_retries": max_retries,
            "retry_count": 0,
            "timeout": timeout,
            "attempts": []
        }

        # Store job metadata
        self.redis_client.hset(
            f"{self.job_prefix}{job_id}",
            mapping=job_data
        )

        # Add to priority queue (lower value = higher priority)
        queue_key = f"{self.queue_prefix}{job_type}"
        self.redis_client.zadd(
            queue_key,
            {job_id: priority.value}
        )

        logger.info(f"Job enqueued: {job_id} (type: {job_type}, priority: {priority.name})")
        return job_id

    def dequeue_job(self, job_type: str, worker_id: str) -> Optional[Dict[str, Any]]:
        """Dequeue a job for processing"""
        queue_key = f"{self.queue_prefix}{job_type}"

        # Get highest priority job (lowest score)
        job_ids = self.redis_client.zrange(queue_key, 0, 0)

        if not job_ids:
            return None

        job_id = job_ids[0]

        # Remove from queue
        self.redis_client.zrem(queue_key, job_id)

        # Get job data
        job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}")

        if not job_data:
            return None

        # Update status and assign to worker
        job_data["status"] = JobStatus.PROCESSING.value
        job_data["worker_id"] = worker_id
        job_data["started_at"] = datetime.utcnow().isoformat()

        self.redis_client.hset(
            f"{self.job_prefix}{job_id}",
            mapping=job_data
        )

        # Track worker activity
        self.redis_client.hset(
            f"{self.worker_prefix}{worker_id}",
            mapping={
                "current_job": job_id,
                "last_heartbeat": datetime.utcnow().isoformat(),
                "jobs_processed": self.redis_client.hget(f"{self.worker_prefix}{worker_id}", "jobs_processed") or 0
            }
        )

        logger.info(f"Job dequeued: {job_id} by worker {worker_id}")
        return job_data

    def complete_job(self, job_id: str, result: Dict[str, Any]) -> bool:
        """Mark job as completed"""
        job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}")

        if not job_data:
            logger.warning(f"Job not found: {job_id}")
            return False

        # Store result
        result_data = {
            "job_id": job_id,
            "status": JobStatus.COMPLETED.value,
            "result": json.dumps(result),
            "completed_at": datetime.utcnow().isoformat()
        }

        self.redis_client.hset(
            f"{self.result_prefix}{job_id}",
            mapping=result_data
        )

        # Update job status
        job_data["status"] = JobStatus.COMPLETED.value
        job_data["completed_at"] = datetime.utcnow().isoformat()

        self.redis_client.hset(
            f"{self.job_prefix}{job_id}",
            mapping=job_data
        )

        # Increment worker stats
        worker_id = job_data.get("worker_id")
        if worker_id:
            self.redis_client.hincrby(
                f"{self.worker_prefix}{worker_id}",
                "jobs_processed",
                1
            )

        logger.info(f"Job completed: {job_id}")
        return True

    def fail_job(self, job_id: str, error: str) -> bool:
        """Handle job failure with retry logic"""
        job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}")

        if not job_data:
            logger.warning(f"Job not found: {job_id}")
            return False

        retry_count = int(job_data.get("retry_count", 0))
        max_retries = int(job_data.get("max_retries", 3))

        # Record attempt
        attempt = {
            "timestamp": datetime.utcnow().isoformat(),
            "error": error,
            "retry_count": retry_count
        }

        attempts = json.loads(job_data.get("attempts", "[]"))
        attempts.append(attempt)

        if retry_count < max_retries:
            # Retry with exponential backoff
            backoff_seconds = min(2 ** retry_count * 60, 3600)  # Max 1 hour
            retry_at = datetime.utcnow() + timedelta(seconds=backoff_seconds)

            job_data["status"] = JobStatus.RETRYING.value
            job_data["retry_count"] = retry_count + 1
            job_data["retry_at"] = retry_at.isoformat()
            job_data["attempts"] = json.dumps(attempts)

            self.redis_client.hset(
                f"{self.job_prefix}{job_id}",
                mapping=job_data
            )

            # Re-queue with delay
            queue_key = f"{self.queue_prefix}{job_data['type']}"
            self.redis_client.zadd(
                queue_key,
                {job_id: int(retry_at.timestamp())}
            )

            logger.info(f"Job retrying: {job_id} (attempt {retry_count + 1}/{max_retries})")
        else:
            # Max retries exceeded
            job_data["status"] = JobStatus.FAILED.value
            job_data["failed_at"] = datetime.utcnow().isoformat()
            job_data["attempts"] = json.dumps(attempts)

            self.redis_client.hset(
                f"{self.job_prefix}{job_id}",
                mapping=job_data
            )

            logger.error(f"Job failed permanently: {job_id} after {max_retries} retries")

        return True

    def get_job_status(self, job_id: str) -> Optional[Dict[str, Any]]:
        """Get current job status"""
        job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}")

        if not job_data:
            return None

        result_data = self.redis_client.hgetall(f"{self.result_prefix}{job_id}")

        return {
            "job": job_data,
            "result": result_data if result_data else None
        }

    def get_queue_stats(self, job_type: str) -> Dict[str, Any]:
        """Get queue statistics"""
        queue_key = f"{self.queue_prefix}{job_type}"
        pending_count = self.redis_client.zcard(queue_key)

        return {
            "job_type": job_type,
            "pending_jobs": pending_count,
            "timestamp": datetime.utcnow().isoformat()
        }

    def get_worker_stats(self, worker_id: str) -> Optional[Dict[str, Any]]:
        """Get worker statistics"""
        worker_data = self.redis_client.hgetall(f"{self.worker_prefix}{worker_id}")

        if not worker_data:
            return None

        return worker_data

    def cleanup_stale_jobs(self, timeout_seconds: int = 3600) -> int:
        """Clean up jobs that have timed out"""
        cursor = 0
        cleaned = 0

        while True:
            cursor, keys = self.redis_client.scan(
                cursor,
                match=f"{self.job_prefix}*",
                count=100
            )

            for key in keys:
                job_data = self.redis_client.hgetall(key)

                if job_data.get("status") == JobStatus.PROCESSING.value:
                    started_at = datetime.fromisoformat(job_data.get("started_at", ""))
                    if datetime.utcnow() - started_at > timedelta(seconds=timeout_seconds):
                        job_id = job_data["id"]
                        self.fail_job(job_id, "Job timeout")
                        cleaned += 1

            if cursor == 0:
                break

        return cleaned

2. Worker Implementation

import time
import signal
import sys
from typing import Callable, Dict, Any
import logging

logger = logging.getLogger(__name__)

class JobWorker:
    """Worker process that executes distributed jobs"""

    def __init__(self, worker_id: str, job_queue: DistributedJobQueue):
        self.worker_id = worker_id
        self.job_queue = job_queue
        self.running = True
        self.job_handlers: Dict[str, Callable] = {}

        # Register signal handlers
        signal.signal(signal.SIGTERM, self._handle_shutdown)
        signal.signal(signal.SIGINT, self._handle_shutdown)

    def register_handler(self, job_type: str, handler: Callable) -> None:
        """Register a handler for a job type"""
        self.job_handlers[job_type] = handler
        logger.info(f"Handler registered for job type: {job_type}")

    def start(self, job_types: list, poll_interval: int = 5) -> None:
        """Start processing jobs"""
        logger.info(f"Worker {self.worker_id} started")

        while self.running:
            job_processed = False

            for job_type in job_types:
                job_data = self.job_queue.dequeue_job(job_type, self.worker_id)

                if job_data:
                    self._process_job(job_data)
                    job_processed = True
                    break

            if not job_processed:
                time.sleep(poll_interval)

    def _process_job(self, job_data: Dict[str, Any]) -> None:
        """Process a single job"""
        job_id = job_data["id"]
        job_type = job_data["type"]

        try:
            logger.info(f"Processing job {job_id} (type: {job_type})")

            # Get handler
            handler = self.job_handlers.get(job_type)
            if not handler:
                raise ValueError(f"No handler for job type: {job_type}")

            # Parse payload
            payload = json.loads(job_data["payload"])

            # Execute job with timeout
            result = self._execute_with_timeout(
                handler,
                payload,
                int(job_data.get("timeout", 3600))
            )

            # Mark as completed
            self.job_queue.complete_job(job_id, result)

        except Exception as e:
            logger.error(f"Job failed: {job_id}, Error: {str(e)}")
            self.job_queue.fail_job(job_id, str(e))

    def _execute_with_timeout(
        self,
        handler: Callable,
        payload: Dict[str, Any],
        timeout: int
    ) -> Dict[str, Any]:
        """Execute handler with timeout"""
        import signal

        def timeout_handler(signum, frame):
            raise TimeoutError(f"Job execution exceeded {timeout} seconds")

        # Set timeout
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout)

        try:
            result = handler(payload)
            signal.alarm(0)  # Cancel alarm
            return result
        except Exception as e:
            signal.alarm(0)
            raise

    def _handle_shutdown(self, signum, frame) -> None:
        """Handle graceful shutdown"""
        logger.info(f"Worker {self.worker_id} shutting down...")
        self.running = False
        sys.exit(0)

3. Example Job Handlers

```python import requests from PIL import Image from io import BytesIO

class OrderProcessingHandlers: """Example job handlers for order processing"""

@staticmethod def process_payment(payload: Dict[str, Any]) -> Dict[str, Any]: """Process payment for an order""" order_id = payload["order_id"] amount = payload["amount"]

logger.info(f"Processing payment for order {order_id}: ${amount}")

Simulate payment processing

time.sleep(2)

return { "order_id": order_id, "status": "paid", "transaction_id": f"TXN-{order_id}-{int(time.time())}" }

@staticmethod def process_images(payload: Dict[str, Any]) -> Dict[str, Any]: """Process product images""" image_urls = payload["image_urls"] order_id = payload["order_id"]

logger.info(f"Processing {len(image_urls)} images for order {order_id}")

processed_images = [] for url in image_urls: try: response = requests.get(url, timeout=10) img = Image.open(BytesIO(response.content))

Resize and optimize

img.thumbnail((800, 800))

processed_images.append({ "original_url": url, "status": "processed" }) except Exception as e: logger.error(f"Failed to process image {url}: {str(e)}") processed_images.append({ "original_url": url, "status": "failed", "error": str(e) })

return { "order_id": order_id, "processed_count": len(processed_images), "images": processed_images }

@staticmethod def send_notification(payload: Dict[str, Any]) -> Dict[str, Any]: """Send email notification""" order_id = payload["order_id"]