Distributed Job Scheduler Design: Avoid the 3 AM PagerDuty Meltdown

// io.thecodeforge — System Design tutorial

// Leader election using etcd lease
// Each node tries to create a key with a TTL. Only one succeeds.

const LEADER_KEY = '/scheduler/leader';
const TTL_SECONDS = 10;

async function tryBecomeLeader(etcdClient, nodeId) {
  // Grant a lease with TTL
  const lease = await etcdClient.lease.grant({ TTL: TTL_SECONDS });
  
  try {
    // Try to create the leader key with the lease. If it exists, this throws.
    await etcdClient.kv.put(LEADER_KEY, nodeId, {
      lease: lease.ID,
      prevKv: false, // ensure key doesn't exist
    });
    console.log(`Node ${nodeId} became leader`);
    return lease;
  } catch (err) {
    if (err.message.includes('key already exists')) {
      console.log(`Node ${nodeId} is follower`);
      lease.revoke(); // don't need the lease
      return null;
    }
    throw err;
  }
}

// Keep lease alive while leader
async function maintainLease(lease) {
  setInterval(async () => {
    await lease.keepAliveOnce();
  }, TTL_SECONDS * 1000 / 2); // renew at half TTL
}

-- io.thecodeforge — System Design tutorial

-- Job table with status and version for optimistic locking
CREATE TABLE jobs (
  id UUID PRIMARY KEY,
  schedule_time TIMESTAMPTZ NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending', -- pending, claimed, running, done, failed
  claimed_by TEXT,
  version INT NOT NULL DEFAULT 1
);

-- Worker claims a job atomically
-- Uses advisory lock to prevent two workers from claiming the same job
UPDATE jobs
SET status = 'claimed',
    claimed_by = 'worker-1',
    version = version + 1
WHERE id = 'job-uuid-here'
  AND status = 'pending'
  AND pg_try_advisory_xact_lock(hashtext(id::text)); -- advisory lock on job ID

-- If 0 rows updated, another worker got it. Retry with another job.

-- io.thecodeforge — System Design tutorial

-- Worker updates heartbeat every 30 seconds
UPDATE jobs
SET last_heartbeat = NOW()
WHERE id = 'job-uuid-here' AND claimed_by = 'worker-1';

-- Watchdog (runs every minute) resets stale jobs
UPDATE jobs
SET status = 'pending',
    claimed_by = NULL,
    version = version + 1
WHERE status = 'claimed'
  AND last_heartbeat < NOW() - INTERVAL '90 seconds'  -- 3x heartbeat interval
  AND pg_try_advisory_xact_lock(hashtext(id::text));

-- Return the count of reset jobs for alerting
SELECT COUNT(*) AS stale_jobs_reset FROM jobs WHERE ... ;

# io.thecodeforge — System Design tutorial

import uuid
from redis import Redis

redis = Redis()

def process_job(job_id: str, run_id: str):
    # Check if this run was already processed
    key = f"job:{job_id}:run:{run_id}"
    if redis.setnx(key, "done"):  # set if not exists, atomic
        # This is the first time we see this run
        # Do the actual work here
        print(f"Processing job {job_id} run {run_id}")
        # ...
        # Set TTL to avoid memory leak
        redis.expire(key, 86400)  # 24 hours
    else:
        print(f"Job {job_id} run {run_id} already processed, skipping")

# Usage
process_job("invoice-batch-123", str(uuid.uuid4()))

# io.thecodeforge — System Design tutorial

import random
from datetime import datetime, timedelta

def schedule_with_jitter(job_definition: dict, now: datetime) -> datetime:
    """
    Returns the actual execution time with jitter.
    job_definition has 'scheduled_time' and 'jitter_window_seconds' (default 60).
    """
    base_time = job_definition['scheduled_time']
    jitter = random.randint(0, job_definition.get('jitter_window_seconds', 60))
    actual_time = base_time + timedelta(seconds=jitter)
    
    # Don't schedule in the past
    if actual_time < now:
        actual_time = now
    return actual_time

# Example: job scheduled for midnight
job = {'scheduled_time': datetime(2025, 1, 1, 0, 0), 'jitter_window_seconds': 120}
print(schedule_with_jitter(job, datetime(2025, 1, 1, 0, 0, 30)))
# Output: 2025-01-01 00:01:45 (random, between 0 and 120 seconds after midnight)

# io.thecodeforge — System Design tutorial

# Prometheus metrics for job scheduler

# Counter: jobs started
job_starts_total = Counter('scheduler_job_starts_total', 'Total number of job starts', ['job_name'])

# Histogram: job duration
job_duration_seconds = Histogram('scheduler_job_duration_seconds', 'Job execution duration', ['job_name'], buckets=[1, 5, 10, 30, 60, 120, 300])

# Gauge: leader status (1 = leader, 0 = follower)
leader_status = Gauge('scheduler_leader_status', '1 if this node is leader, 0 otherwise')

# In job execution:
job_starts_total.labels(job_name=job.name).inc()
with job_duration_seconds.labels(job_name=job.name).time():
    # run job
    pass

# In leader election:
leader_status.set(1 if is_leader else 0)