background-jobs

name	background-jobs
description	Background job processing patterns including job queues, scheduled jobs, worker pools, and retry strategies. Use when implementing async processing, Celery, Bull, Sidekiq, cron jobs, task queues, job monitoring, or worker management.

Background Jobs

Overview

Background jobs enable asynchronous processing of tasks outside the request-response cycle. This skill covers job queue patterns, scheduling, worker management, retry strategies, and monitoring for reliable task execution across different frameworks and languages.

Key Concepts

Job Queue Patterns

Bull Queue (Node.js/Redis):

import Queue, { Job, JobOptions } from "bull";
import { Redis } from "ioredis";

// Queue configuration
interface QueueConfig {
  name: string;
  redis: Redis;
  defaultJobOptions?: JobOptions;
}

// Job data interfaces
interface EmailJobData {
  to: string;
  subject: string;
  template: string;
  context: Record<string, unknown>;
}

interface ImageProcessingJobData {
  imageId: string;
  operations: Array<{
    type: "resize" | "crop" | "compress";
    params: Record<string, unknown>;
  }>;
}

// Queue factory
function createQueue<T>(config: QueueConfig): Queue.Queue<T> {
  const queue = new Queue<T>(config.name, {
    createClient: (type) => {
      switch (type) {
        case "client":
          return config.redis.duplicate();
        case "subscriber":
          return config.redis.duplicate();
        case "bclient":
          return config.redis.duplicate();
        default:
          return config.redis.duplicate();
      }
    },
    defaultJobOptions: {
      removeOnComplete: 100, // Keep last 100 completed jobs
      removeOnFail: 1000, // Keep last 1000 failed jobs
      attempts: 3,
      backoff: {
        type: "exponential",
        delay: 2000,
      },
      ...config.defaultJobOptions,
    },
  });

  // Global error handler
  queue.on("error", (error) => {
    console.error(`Queue ${config.name} error:`, error);
  });

  return queue;
}

// Email queue with typed processor
const emailQueue = createQueue<EmailJobData>({
  name: "email",
  redis: new Redis(process.env.REDIS_URL),
});

// Define processor
emailQueue.process(async (job: Job<EmailJobData>) => {
  const { to, subject, template, context } = job.data;

  // Update progress
  await job.progress(10);

  // Render template
  const html = await renderTemplate(template, context);
  await job.progress(50);

  // Send email
  await emailService.send({ to, subject, html });
  await job.progress(100);

  return { sent: true, messageId: `msg_${Date.now()}` };
});

// Add job with options
async function sendEmail(
  data: EmailJobData,
  options?: JobOptions,
): Promise<Job<EmailJobData>> {
  return emailQueue.add(data, {
    priority: options?.priority || 0,
    delay: options?.delay || 0,
    jobId: options?.jobId, // For deduplication
    ...options,
  });
}

// Bulk job addition
async function sendBulkEmails(
  emails: EmailJobData[],
): Promise<Job<EmailJobData>[]> {
  const jobs = emails.map((data, index) => ({
    data,
    opts: {
      jobId: `bulk_${Date.now()}_${index}`,
    },
  }));

  return emailQueue.addBulk(jobs);
}

Celery (Python):

from celery import Celery, Task
from celery.exceptions import MaxRetriesExceededError
from typing import Any, Dict, Optional
import logging

# Celery configuration
app = Celery('tasks')
app.config_from_object({
    'broker_url': 'redis://localhost:6379/0',
    'result_backend': 'redis://localhost:6379/1',
    'task_serializer': 'json',
    'result_serializer': 'json',
    'accept_content': ['json'],
    'timezone': 'UTC',
    'task_track_started': True,
    'task_time_limit': 300,  # 5 minutes hard limit
    'task_soft_time_limit': 240,  # 4 minutes soft limit
    'worker_prefetch_multiplier': 4,
    'task_acks_late': True,  # Acknowledge after task completes
    'task_reject_on_worker_lost': True,
})

logger = logging.getLogger(__name__)

# Base task with retry logic
class BaseTask(Task):
    autoretry_for = (Exception,)
    retry_kwargs = {'max_retries': 3}
    retry_backoff = True
    retry_backoff_max = 600  
    retry_jitter = 

     ():
        logger.error()

     ():
        logger.warning()

     ():
        logger.info()



 () -> [, ]:
    :
        
        .update_state(state=, meta={: })

        
        html = render_template(template, context)
        .update_state(state=, meta={: })

        
        message_id = email_service.send(to=to, subject=subject, html=html)
        .update_state(state=, meta={: })

         {: , : message_id}

     ConnectionError  exc:
         .retry(exc=exc, countdown=)



 () -> [, ]:
    image = load_image(image_id)

     i, op  (operations):
        progress = ((i + ) / (operations) * )
        .update_state(state=, meta={: progress, : op[]})

         op[] == :
            image = resize_image(image, **op[])
         op[] == :
            image = crop_image(image, **op[])
         op[] == :
            image = compress_image(image, **op[])

    url = save_image(image, image_id)
     {: url, : (operations)}


 celery  chain, group, chord

 ():
    
    workflow = chain(
        validate_order.s(order_id),
        reserve_inventory.s(),
        process_payment.s(),
        send_confirmation.s(),
    )
     workflow.apply_async()

 ():
    
    workflow = chord(
        group(process_image.s(img_id, [{: , : {: }}])
               img_id  image_ids),
        aggregate_results.s()
    )
     workflow.apply_async()

Sidekiq (Ruby):

# config/initializers/sidekiq.rb
Sidekiq.configure_server do |config|
  config.redis = { url: ENV['REDIS_URL'], network_timeout: 5 }
  config.death_handlers << ->(job, ex) do
    # Handle job failure
    ErrorReporter.report(ex, job: job)
  end
end

Sidekiq.configure_client do |config|
  config.redis = { url: ENV['REDIS_URL'], network_timeout: 5 }
end

# app/workers/email_worker.rb
class EmailWorker
  include Sidekiq::Worker

  sidekiq_options queue: :default,
                  retry: 5,
                  backtrace: true,
                  dead: true

  sidekiq_retry_in do |count, exception|
    # Exponential backoff: 1, 8, 27, 64, 125 seconds
    (count + 1) ** 3
  end

  sidekiq_retries_exhausted do |msg, exception|
    Rails.logger.error "Job #{msg['jid']} exhausted retries: "
    .notify(msg, exception)
  

   ()
    html = .render(
       template,
       context.symbolize_keys
    )

    .send( to,  subject,  html)
  



 
   

   ()
    batch = .find(batch_id)

    batch.items.find_each  ||
      .perform_async(item.id)
    
  



 
   

   ()
    import = .find(import_id)

    batch = .new
    batch.description = 
    batch.on(, ,  import_id)

    batch.jobs 
      import.rows.each_with_index  ||
        .perform_async(import_id, index, row)
      
    
  


 
   ()
    import = .find(options[])

     status.failures.zero?
      import.update!( )
    
      import.update!( ,  status.failures)
    
  

Scheduled Jobs and Cron Patterns

// Bull scheduler
import Queue from "bull";

const scheduledQueue = new Queue("scheduled-tasks", process.env.REDIS_URL);

// Repeatable jobs
async function setupScheduledJobs(): Promise<void> {
  // Clean up every hour
  await scheduledQueue.add(
    "cleanup",
    {},
    {
      repeat: { cron: "0 * * * *" }, // Every hour
      jobId: "cleanup-hourly",
    },
  );

  // Daily report at 9 AM
  await scheduledQueue.add(
    "daily-report",
    {},
    {
      repeat: { cron: "0 9 * * *" },
      jobId: "daily-report",
    },
  );

  // Every 5 minutes
  await scheduledQueue.add(
    "health-check",
    {},
    {
      repeat: { every: 5 * 60 * 1000 }, // 5 minutes in ms
      jobId: "health-check",
    },
  );

  // Weekly on Sunday at midnight
   scheduledQueue.(
    ,
    {},
    {
      : { :  },
      : ,
    },
  );
}


scheduledQueue.(,  (job) => {
   ();
   { :  };
});

scheduledQueue.(,  (job) => {
   report =  ();
   (report);
   { : report. };
});


  (): <
  <{ : ; : ; :  }>
> {
   repeatableJobs =  scheduledQueue.();

   repeatableJobs.( ({
    : job.,
    :  (job.),
    : job. || ,
  }));
}


  (): <> {
   jobs =  scheduledQueue.();
   job = jobs.( j. === jobId);

   (job) {
     scheduledQueue.(job.);
  }
}

# Celery Beat scheduler
from celery import Celery
from celery.schedules import crontab

app = Celery('tasks')

app.conf.beat_schedule = {
    # Every hour
    'cleanup-hourly': {
        'task': 'tasks.cleanup',
        'schedule': crontab(minute=0),  # Every hour at minute 0
    },

    # Daily at 9 AM
    'daily-report': {
        'task': 'tasks.daily_report',
        'schedule': crontab(hour=9, minute=0),
    },

    # Every 5 minutes
    'health-check': {
        'task': 'tasks.health_check',
        'schedule': 300.0,  # 5 minutes in seconds
    },

    # Weekly on Sunday at midnight
    'weekly-cleanup': {
        'task': 'tasks.weekly_cleanup',
        'schedule': crontab(hour=0, minute=0, day_of_week=0),
    },

    # First day of month at 6 AM
    'monthly-report': {
        'task': 'tasks.monthly_report',
        'schedule': crontab(hour=6, minute=0, day_of_month=1),
    },

    # With arguments
    'check-expiring-subscriptions': {
        'task': ,
        : crontab(hour=, minute=),
        : (,),
        : {: },
    },
}


 django_celery_beat.models  PeriodicTask, CrontabSchedule
 json

 ():
    
    
    minute, hour, day_of_month, month, day_of_week = cron.split()

    schedule, _ = CrontabSchedule.objects.get_or_create(
        minute=minute,
        hour=hour,
        day_of_month=day_of_month,
        month_of_year=month,
        day_of_week=day_of_week,
    )

    PeriodicTask.objects.update_or_create(
        name=name,
        defaults={
            : task,
            : schedule,
            : json.dumps(args  []),
            : json.dumps(kwargs  {}),
            : ,
        },
    )

Worker Pool Management

import Queue, { Job } from "bull";
import os from "os";

interface WorkerPoolConfig {
  concurrency: number;
  limiter?: {
    max: number;
    duration: number;
  };
}

class WorkerPool {
  private queues: Map<string, Queue.Queue> = new Map();
  private isShuttingDown = false;

  constructor(private config: WorkerPoolConfig) {
    // Graceful shutdown
    process.on("SIGTERM", () => this.shutdown());
    process.on("SIGINT", () => this.shutdown());
  }

  registerQueue<T>(
    name: string,
    processor: (job: Job<T>) => Promise<>,
  ): .<T> {
     queue =  <T>(name, process..!, {
      : ..,
    });

    
    queue.(..,  (: <T>) => {
       (.) {
          ();
      }
       (job);
    });

    
    queue.(,  {
      .(, result);
    });

    queue.(,  {
      .(, err);
    });

    queue.(,  {
      .();
    });

    ..(name, queue);
     queue;
  }

   (): <> {
    .();
    . = ;

    
     closePromises = .(..()).(
       (queue) => {
         queue.(); 
         queue.();
      },
    );

     .(closePromises);
    .();
    process.();
  }

   (): <<, >> {
     : <, > = {};

     ( [name, queue]  .) {
       [waiting, active, completed, failed, delayed] =  .([
        queue.(),
        queue.(),
        queue.(),
        queue.(),
        queue.(),
      ]);

      stats[name] = { waiting, active, completed, failed, delayed };
    }

     stats;
  }
}

  {
  : ;
  : ;
  : ;
  : ;
  : ;
}


 pool =  ({
  : os.().,
  : {
    : , 
    : , 
  },
});

pool.<>(,  (job) => {
   (job.);
});

pool.<>(,  (job) => {
   (job.);
});

# Celery worker management
from celery import Celery
from celery.signals import worker_process_init, worker_shutdown
import multiprocessing

app = Celery('tasks')

# Worker configuration
app.conf.update(
    worker_concurrency=multiprocessing.cpu_count(),
    worker_prefetch_multiplier=2,  # Prefetch 2 tasks per worker
    worker_max_tasks_per_child=1000,  # Restart worker after 1000 tasks
    worker_max_memory_per_child=200000,  # 200MB limit
    task_acks_late=True,
    task_reject_on_worker_lost=True,
)

# Per-worker initialization
@worker_process_init.connect
def init_worker(**kwargs):
    """Initialize resources for each worker process."""
    # Initialize database connection pool
    db.connect()
    # Warm up caches
    cache.warm_up()

@worker_shutdown.connect
def cleanup_worker(**kwargs):
    """Clean up resources on worker shutdown."""
    db.close()
    cache.flush()

# Task routing for specialized workers
app.conf.task_routes = {
    'tasks.send_email': {'queue': 'email'},
    'tasks.process_image': {'queue': 'images'},
    'tasks.heavy_computation': {'queue': 'compute'},
    'tasks.*': {'queue': 'default'},
}

# Queue-specific worker command:





app.conf.worker_autoscaler = 
app.conf.worker_autoscale_max = 
app.conf.worker_autoscale_min = 

Job Priorities and Fairness

// Priority queues with Bull
interface PriorityJobData {
  type: string;
  payload: unknown;
  priority: "critical" | "high" | "normal" | "low";
}

const priorityMap = {
  critical: 1, // Highest priority (processed first)
  high: 5,
  normal: 10,
  low: 20,
};

async function addPriorityJob(
  data: PriorityJobData,
): Promise<Job<PriorityJobData>> {
  return queue.add(data, {
    priority: priorityMap[data.priority],
    // Critical jobs don't wait
    delay: data.priority === "critical" ? 0 : undefined,
  });
}

// Fair scheduling with multiple queues
class FairScheduler {
  private queues: Map<string, Queue.Queue> = new Map();
   : <, > =  ();

  () {
     ( config  queueConfigs) {
       queue =  (config., process..!);
      ..(config., queue);
      ..(config., config.);
    }
  }

  
   (
    :  <>,
  ): <> {
     totalWeight = .(..()).(
       a + b,
      ,
    );

     ( [name, queue]  .) {
       weight = ..(name)!;
       concurrency = .(, .((weight / totalWeight) * ));

      queue.(concurrency,  (job) => {
         (name, job);
      });
    }
  }
}


 scheduler =  ([
  { : , :  }, 
  { : , :  }, 
  { : , :  }, 
]);

 scheduler.( (queueName, job) => {
  .(, job.);
   (job);
});

Idempotency and Retry Strategies

import Queue, { Job, JobOptions } from "bull";
import { createHash } from "crypto";

// Idempotency key generation
function generateIdempotencyKey(data: unknown): string {
  const hash = createHash("sha256");
  hash.update(JSON.stringify(data));
  return hash.digest("hex");
}

// Idempotent job processor
class IdempotentProcessor<T> {
  private processedKeys: Set<string> = new Set();
  private redis: Redis;

  constructor(
    private queue: Queue.Queue<T>,
    redis: Redis,
  ) {
    this.redis = redis;
  }

  async process(handler: (job: <T>) => <>): <> {
    ..( (: <T>) => {
       idempotencyKey = job.. || (job.);

      
       existing =  ..();
       (existing) {
        .();
         .(existing);
      }

      
       result =  (job);

      
       ..(
        ,
        , 
        .(result),
      );

       result;
    });
  }
}


  {
  :  |  |  | ;
  : ;
  ?: ;
  : ;
  ?: ;
  ?:  ;
}

 ():  {
   : ;

   (strategy.) {
     :
      delay = strategy. * .(, attempt - );
      ;
     :
      delay = strategy. * attempt;
      ;
     :
      delay = strategy.;
      ;
    :
      delay = strategy.;
  }

  
   (strategy.) {
    delay = .(delay, strategy.);
  }

  
   (strategy.) {
     jitterFactor =  + .() * ; 
    delay = .(delay * jitterFactor);
  }

   delay;
}


 <T> {
   : .<T>;
   : .<T>;
   : ;

  () {
    . =  <T>(name, process..!);
    . =  <T>(, process..!);
    . = strategy;
  }

   (:  <>): <> {
    ..( (: <T>) => {
       attempts = job.;

       {
          (job);
      }  (error) {
         err = error  ;

        
         (.. && !..(err)) {
           .(job, err);
           err;
        }

        
         (attempts >= ..) {
           .(job, err);
           err;
        }

        
         delay = (., attempts + );
          ();
      }
    });
  }

    (: <T>, : ): <> {
     ..({
      : job.,
      : error.,
      :  ().(),
      : job.,
    }    T);
  }

   (: ): <> {
     job =  ..(jobId);
     (!job) ;

     dlqData = job.    { : T };
     ..(dlqData.);
     job.();
  }
}

Job Monitoring and Dead Jobs

import Queue, { Job, JobCounts, JobStatus } from "bull";
import { EventEmitter } from "events";

interface JobMetrics {
  queue: string;
  counts: JobCounts;
  latency: {
    avg: number;
    p50: number;
    p95: number;
    p99: number;
  };
  throughput: number; // jobs per minute
  errorRate: number;
}

class JobMonitor extends EventEmitter {
  private queues: Queue.Queue[] = [];
  private metricsHistory: Map<string, number[]> = new Map();

  addQueue(queue: Queue.Queue): void {
    this.queues.push(queue);

    queue.(,  {
      .(queue., , job);
      .(, { : queue., job, result });
    });

    queue.(,  {
      .(queue., , job!);
      .(, { : queue., job, : err });

      
      .(queue.);
    });

    queue.(,  {
      .(, { : queue., : job });
    });
  }

   (: , : , : ):  {
     duration = .() - job.;
     key = ;

     history = ..(key) || [];
    history.(duration);

    
     (history. > ) {
      history.();
    }

    ..(key, history);
  }

   (: ):  {
     completed =
      ..()?. || ;
     failed =
      ..()?. || ;

     (completed + failed > ) {
       errorRate = failed / (completed + failed);
       (errorRate > ) {
        
        .(, { : queueName, errorRate });
      }
    }
  }

   (: ): <> {
     queue = ..( q. === queueName);
     (!queue)   ();

     counts =  queue.();
     durations =
      ..() || [];

     {
      : queueName,
      counts,
      : .(durations),
      : .(durations),
      : .(queueName),
    };
  }

   (
    : [],
  ): [] {
     (durations. === ) {
       { : , : , : , :  };
    }

     sorted = [...durations].( a - b);
     avg = sorted.( a + b, ) / sorted.;

     {
      : .(avg),
      : sorted[.(sorted. * )],
      : sorted[.(sorted. * )],
      : sorted[.(sorted. * )],
    };
  }

   (: []):  {
    
     oneMinuteAgo = .() - ;
     recentJobs = durations.( i > durations. - );
     recentJobs.;
  }

   (: ):  {
     completed =
      ..()?. || ;
     failed =
      ..()?. || ;
     total = completed + failed;
     total >  ? failed / total : ;
  }

  
   (: , :  = ): <[]> {
     queue = ..( q. === queueName);
     (!queue)   ();

     queue.(, limit);
  }

   (: , : ): <> {
     queue = ..( q. === queueName);
     (!queue)   ();

     job =  queue.(jobId);
     (!job)   ();

     job.();
  }

   (: ): <> {
     deadJobs =  .(queueName);
     retried = ;

     ( job  deadJobs) {
       {
         job.();
        retried++;
      }  (error) {
        .(, error);
      }
    }

     retried;
  }

   (
    : ,
    :  = ,
  ): <> {
     queue = ..( q. === queueName);
     (!queue)   ();

     cleaned =  queue.(olderThan, );
     cleaned.;
  }
}


 express  ;

 (): express. {
   router = express.();

  router.(,  (req, res) => {
     {
       metrics =  monitor.(req..);
      res.(metrics);
    }  (error) {
      res.().({ : (error  ). });
    }
  });

  router.(,  (req, res) => {
     limit = (req..  ) || ;
     jobs =  monitor.(req.., limit);
    res.(
      jobs.( ({
        : j.,
        : j.,
        : j.,
        : j.,
        : j.,
      })),
    );
  });

  router.(,  (req, res) => {
     {
       monitor.(req.., req..);
      res.({ :  });
    }  (error) {
      res.().({ : (error  ). });
    }
  });

  router.(,  (req, res) => {
     retried =  monitor.(req..);
    res.({ retried });
  });

  router.(,  (req, res) => {
     olderThan = (req..  ) || ;
     cleaned =  monitor.(req.., olderThan);
    res.({ cleaned });
  });

   router;
}

Best Practices

1. Idempotency

   • Design jobs to be safely re-executed
   • Use unique job IDs for deduplication
   • Store processed state externally

2. Retry Strategies

   • Use exponential backoff with jitter
   • Set maximum retry limits
   • Distinguish between retryable and non-retryable errors

3. Monitoring

   • Track queue depths and processing latency
   • Alert on high error rates or growing queues
   • Monitor worker health and memory usage

4. Graceful Shutdown

   • Complete in-progress jobs before shutdown
   • Use signals (SIGTERM, SIGINT) properly
   • Set reasonable timeouts for job completion

5. Resource Management

   • Set appropriate concurrency limits
   • Use worker pools for CPU-bound tasks
   • Implement rate limiting for external APIs

Examples

Complete Worker Service

import Queue, { Job } from "bull";
import { Redis } from "ioredis";

interface WorkerConfig {
  queues: Array<{
    name: string;
    concurrency: number;
    processor: (job: Job) => Promise<unknown>;
  }>;
  redis: Redis;
  shutdownTimeout: number;
}

class WorkerService {
  private queues: Map<string, Queue.Queue> = new Map();
  private isShuttingDown = false;
  private activeJobs = 0;

  constructor(private config: WorkerConfig) {}

  async start(): Promise<void> {
    // Setup queues
    for (const queueConfig  ..) {
       queue =  (queueConfig., {
        :  ...(),
      });

      queue.(queueConfig.,  (job) => {
         (.) {
            ();
        }

        .++;
         {
            queueConfig.(job);
        }  {
          .--;
        }
      });

      ..(queueConfig., queue);
    }

    
    process.(,  .());
    process.(,  .());

    .();
  }

    (): <> {
     (.) ;
    . = ;

    .();

    
     .(
      .(..()).( q.()),
    );

    
     startTime = .();
     (
      . >  &&
      .() - startTime < ..
    ) {
        ( (r, ));
    }

     (. > ) {
      .();
    }

    
     .(.(..()).( q.()));

    .();
    process.();
  }
}


 worker =  ({
  :  (process..),
  : ,
  : [
    {
      : ,
      : ,
      :  (job) => {
         (job.);
      },
    },
    {
      : ,
      : ,
      :  (job) => {
         (job.);
      },
    },
  ],
});

worker.();