Go Guide

System design · Lesson 62 of 62 · about 90 min

Design a distributed job scheduler

Run background and scheduled jobs reliably across many workers: delayed and recurring jobs, leases with heartbeats so a crashed worker's job is picked up again, fencing out stale workers, retries with backoff, and SKIP LOCKED queues, then build a job scheduler in Go with a live dashboard.

PostgreSQL: SELECT ... FOR UPDATE SKIP LOCKED ↗Martin Kleppmann: How to do distributed locking (fencing tokens) ↗Amazon SQS: Visibility timeout ↗
On this page · 9 sections
  1. 011. Requirements
  2. 022. Estimates
  3. 033. Leases, heartbeats and fencing
  4. 044. High-level design
  5. 055. Deep dives
  6. 06Build it: a job scheduler in Go
  7. 07Summary
  8. 08Check your understanding
  9. 09Practice in Go

1. Requirements#

Functional: enqueue a job to run now, at a time, or on a schedule (every N minutes, or cron); workers execute jobs by kind; failed jobs retry with backoff up to a limit, then land in a dead state for inspection; jobs and their history are visible.

Non-functional: no job is lost, even if a worker dies mid-run; a job does not run on two workers at once (outside of crash recovery); scheduled jobs start within a second or two of their time; it scales to many workers and millions of queued jobs.

2. Estimates#

10 million jobs a day is about 115 per second on average. Peaks come from schedules: thousands of "every hour" jobs all fire at minute zero. If a job averages 2 seconds of work, 115 jobs per second needs about 230 busy workers on average, and more at the top of the hour. Spreading recurring jobs (adding a random offset to each schedule) flattens those peaks for free.

3. Leases, heartbeats and fencing#

Drawing diagram…
  • Claim: atomically pick a due job and set state = running, lease_until = now + 30s, worker = A, attempt = attempt + 1.
  • Heartbeat: every 10 seconds, extend lease_until. The lease must be longer than the heartbeat interval with room for hiccups.
  • Recovery: a running job whose lease has passed is claimable again. No coordinator or failure detector is needed; the timestamp says it all.
  • Fencing: every heartbeat and completion carries the lease (worker + attempt). A worker that paused and came back with an old lease is rejected, so it cannot overwrite the result of the worker that took over.
lease.go
Go
package main
 
import (
	"fmt"
	"time"
)
 
type job struct {
	worker     string
	attempt    int
	leaseUntil time.Time
	done       bool
}
 
func (j *job) claim(worker string, now time.Time) bool {
	if j.done || (j.worker != "" && now.Before(j.leaseUntil)) {
		return false // someone holds a live lease
	}
	j.worker, j.attempt, j.leaseUntil = worker, j.attempt+1, now.Add(30*time.Second)
	return true
}
 
func (j *job) complete(worker string, attempt int) error {
	if j.worker != worker || j.attempt != attempt {
		return fmt.Errorf("%s attempt %d: lease lost", worker, attempt)
	}
	j.done = true
	return nil
}
 
func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	j := &job{}
	fmt.Println("A claims:", j.claim("A", t0))
	fmt.Println("B claims 10s later:", j.claim("B", t0.Add(10*time.Second)))
	// A freezes. Its lease runs out at 12:00:30.
	fmt.Println("B claims 40s later:", j.claim("B", t0.Add(40*time.Second)), "attempt", j.attempt)
	fmt.Println("A wakes up:", j.complete("A", 1))
	fmt.Println("B finishes:", j.complete("B", 2))
}
Output
A claims: true
B claims 10s later: false
B claims 40s later: true attempt 2
A wakes up: A attempt 1: lease lost
B finishes: <nil>

4. High-level design#

Drawing diagram…

A single table indexed on (state, run_at) goes a long way. Postgres's FOR UPDATE SKIP LOCKED lets many workers claim different rows at once without blocking each other:

SQL
UPDATE jobs SET state = 'running', attempt = attempt + 1, worker = $1, lease_until = now() + interval '30 seconds'
WHERE id = (
  SELECT id FROM jobs
  WHERE (state = 'pending' AND run_at <= now()) OR (state = 'running' AND lease_until < now())
  ORDER BY run_at
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
RETURNING *;

At larger scale, a timer service moves due jobs into a message queue (Kafka, SQS) and workers consume from it. SQS's visibility timeout is exactly a lease: a received message is hidden for N seconds and reappears if it is not deleted.

5. Deep dives#

  • At-least-once, so idempotent jobs. A worker can finish the work and crash before it reports success; the job runs again. Make jobs idempotent: "charge order 881 with key order-881", not "charge $20".
  • Recurring jobs. When a recurring run completes, compute the next time from the schedule, not from "now", so runs do not drift. If the system was down for three periods, run once and skip the missed ones rather than firing three times in a row.
  • Retries. Back off exponentially, cap the attempts, and keep failed jobs in a dead state with the last error. A dashboard button to retry them is the most-used feature in practice.
  • Priorities and fairness. Separate queues (or a priority column) keep urgent jobs ahead of batch work; per-tenant limits stop one customer's 1 million jobs from starving everyone else.
  • Exactly-once scheduling of recurring jobs. If several scheduler instances create "nightly report" jobs, they need a unique key per run (report:2026-01-01) so only one instance's insert succeeds.

Build it: a job scheduler in Go#

You will build a jobs package with a store that supports delayed, recurring and retried jobs, leases with heartbeats and fencing, and a runner that executes handlers. All lease logic is tested with a fake clock. The demo starts three workers and a dashboard where you can enqueue jobs that succeed, fail, run long, or "crash" their worker, and watch them recover.

scheduler/
├── go.mod
├── main.go                       API, handlers, three workers
├── main_test.go
├── internal/jobs/store.go        claim, heartbeat, complete, schedules
├── internal/jobs/runner.go       worker loop with heartbeats
├── internal/jobs/jobs_test.go
└── web/index.html                live dashboard

Step 1: module

Shell
mkdir scheduler && cd scheduler && go mod init scheduler
mkdir -p internal/jobs web
go.mod
module scheduler
 
go 1.24

Step 2: the store

The store is in memory behind one mutex, but every method maps to one SQL statement in the Postgres version. Claim returns a Lease, and every later call must present it.

internal/jobs/store.go
Go
// Package jobs schedules and runs background jobs with leases, heartbeats and retries.
package jobs
 
import (
	"errors"
	"fmt"
	"slices"
	"strings"
	"sync"
	"time"
)
 
// State of a job.
type State string
 
const (
	Pending   State = "pending"
	Running   State = "running"
	Succeeded State = "succeeded"
	Dead      State = "dead" // failed permanently or ran out of attempts
)
 
// Job is one unit of work. Every > 0 makes it recurring.
type Job struct {
	ID          string        `json:"id"`
	Kind        string        `json:"kind"`
	Payload     string        `json:"payload"`
	State       State         `json:"state"`
	RunAt       time.Time     `json:"run_at"`
	Every       time.Duration `json:"every"`
	Attempts    int           `json:"attempts"` // attempts of the current run
	MaxAttempts int           `json:"max_attempts"`
	Runs        int           `json:"runs"` // completed successful runs
	Worker      string        `json:"worker,omitempty"`
	LeaseUntil  time.Time     `json:"lease_until"`
	LastError   string        `json:"last_error,omitempty"`
}
 
// Spec describes a job to enqueue.
type Spec struct {
	Kind        string
	Payload     string
	RunAt       time.Time // zero means now
	Every       time.Duration
	MaxAttempts int // zero means 3
}
 
// Lease proves ownership of a running job. It is checked on every heartbeat and completion.
type Lease struct {
	JobID   string
	Worker  string
	Attempt int
}
 
var (
	ErrLeaseLost = errors.New("lease lost")
	ErrNotFound  = errors.New("job not found")
	// ErrPermanent wrapped in a handler's error means retrying cannot help.
	ErrPermanent = errors.New("permanent failure")
)
 
// Store holds jobs. Lease is how long a claim lasts without a heartbeat;
// Backoff is the wait before the first retry, doubled for each one after.
type Store struct {
	mu      sync.Mutex
	jobs    map[string]*Job
	seq     int
	lease   time.Duration
	backoff time.Duration
	now     func() time.Time
}
 
func NewStore(lease, backoff time.Duration, now func() time.Time) *Store {
	if now == nil {
		now = time.Now
	}
	return &Store{jobs: map[string]*Job{}, lease: lease, backoff: backoff, now: now}
}
 
// Enqueue adds a job and returns a copy of it.
func (s *Store) Enqueue(spec Spec) Job {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.seq++
	j := &Job{
		ID: fmt.Sprintf("job-%04d", s.seq), Kind: spec.Kind, Payload: spec.Payload, State: Pending,
		RunAt: spec.RunAt, Every: spec.Every, MaxAttempts: spec.MaxAttempts,
	}
	if j.RunAt.IsZero() {
		j.RunAt = s.now()
	}
	if j.MaxAttempts <= 0 {
		j.MaxAttempts = 3
	}
	s.jobs[j.ID] = j
	return *j
}
 
// Claim gives worker the due job with the earliest run time: a pending job whose time has
// come, or a running job whose lease expired (its worker is presumed dead).
func (s *Store) Claim(worker string) (Lease, Job, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := s.now()
	var best *Job
	for _, j := range s.jobs {
		due := j.State == Pending && !j.RunAt.After(now)
		expired := j.State == Running && now.After(j.LeaseUntil)
		if expired && j.Attempts >= j.MaxAttempts {
			j.State, j.LastError, j.Worker = Dead, "lease expired on the last attempt", ""
			continue
		}
		if (due || expired) && (best == nil || j.RunAt.Before(best.RunAt) || (j.RunAt.Equal(best.RunAt) && j.ID < best.ID)) {
			best = j
		}
	}
	if best == nil {
		return Lease{}, Job{}, false
	}
	best.State, best.Worker = Running, worker
	best.Attempts++
	best.LeaseUntil = now.Add(s.lease)
	return Lease{best.ID, worker, best.Attempts}, *best, true
}
 
// ownedLocked returns the job if the lease still owns it.
func (s *Store) ownedLocked(l Lease) (*Job, error) {
	j, ok := s.jobs[l.JobID]
	if !ok {
		return nil, ErrNotFound
	}
	if j.State != Running || j.Worker != l.Worker || j.Attempts != l.Attempt {
		return nil, ErrLeaseLost
	}
	return j, nil
}
 
// Heartbeat extends the lease.
func (s *Store) Heartbeat(l Lease) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	j, err := s.ownedLocked(l)
	if err != nil {
		return err
	}
	j.LeaseUntil = s.now().Add(s.lease)
	return nil
}
 
// Complete records the outcome of an attempt: success, a retry with backoff, or dead.
func (s *Store) Complete(l Lease, runErr error) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	j, err := s.ownedLocked(l)
	if err != nil {
		return err
	}
	now := s.now()
	j.Worker, j.LeaseUntil = "", time.Time{}
	switch {
	case runErr == nil:
		j.Runs++
		j.LastError = ""
		if j.Every > 0 {
			// Next run from the schedule, not from now, skipping runs missed while down.
			next := j.RunAt.Add(j.Every)
			for !next.After(now) {
				next = next.Add(j.Every)
			}
			j.State, j.RunAt, j.Attempts = Pending, next, 0
		} else {
			j.State = Succeeded
		}
	case errors.Is(runErr, ErrPermanent) || j.Attempts >= j.MaxAttempts:
		j.State, j.LastError = Dead, runErr.Error()
	default:
		j.State, j.LastError = Pending, runErr.Error()
		j.RunAt = now.Add(s.backoff << (j.Attempts - 1))
	}
	return nil
}
 
// Retry moves a dead job back to pending with fresh attempts, as a dashboard button would.
func (s *Store) Retry(id string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	j, ok := s.jobs[id]
	if !ok {
		return ErrNotFound
	}
	if j.State != Dead {
		return fmt.Errorf("%s is %s, not dead", id, j.State)
	}
	j.State, j.Attempts, j.RunAt = Pending, 0, s.now()
	return nil
}
 
// List returns every job, ordered by ID.
func (s *Store) List() []Job {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Job, 0, len(s.jobs))
	for _, j := range s.jobs {
		out = append(out, *j)
	}
	slices.SortFunc(out, func(a, b Job) int { return strings.Compare(a.ID, b.ID) })
	return out
}

Step 3: the runner

A runner is one worker: claim, run the handler while a goroutine heartbeats, then complete. If a heartbeat is rejected, the lease is gone, so the runner cancels the handler's context and does not report a result. ErrAbandon lets the demo simulate a worker that dies mid-job: no more heartbeats, no completion.

internal/jobs/runner.go
Go
package jobs
 
import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"time"
)
 
// Handler runs one job. It should stop when ctx is cancelled and be idempotent.
type Handler func(ctx context.Context, j Job) error
 
// ErrAbandon, returned by a handler, simulates a worker crash: the runner walks away
// without completing, and the lease expires.
var ErrAbandon = errors.New("worker crashed")
 
// Runner is one worker.
type Runner struct {
	Store     *Store
	Name      string
	Handlers  map[string]Handler
	Poll      time.Duration // wait when there is nothing to do
	Heartbeat time.Duration // must be well under the store's lease
}
 
// Run claims and executes jobs until ctx is cancelled.
func (r *Runner) Run(ctx context.Context) {
	for ctx.Err() == nil {
		lease, job, ok := r.Store.Claim(r.Name)
		if !ok {
			select {
			case <-ctx.Done():
			case <-time.After(r.Poll):
			}
			continue
		}
		r.runOne(ctx, lease, job)
	}
}
 
func (r *Runner) runOne(ctx context.Context, lease Lease, job Job) {
	h, ok := r.Handlers[job.Kind]
	if !ok {
		r.Store.Complete(lease, fmt.Errorf("no handler for kind %q: %w", job.Kind, ErrPermanent))
		return
	}
	jobCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	stop := make(chan struct{})
	lost := make(chan struct{})
	go func() {
		t := time.NewTicker(r.Heartbeat)
		defer t.Stop()
		for {
			select {
			case <-stop:
				return
			case <-t.C:
				if err := r.Store.Heartbeat(lease); err != nil {
					close(lost)
					cancel() // someone else owns the job now: stop working on it
					return
				}
			}
		}
	}()
	err := h(jobCtx, job)
	close(stop)
	select {
	case <-lost:
		slog.Warn("lease lost, result discarded", "job", job.ID, "worker", r.Name)
		return
	default:
	}
	if errors.Is(err, ErrAbandon) {
		slog.Warn("worker crashed mid-job (simulated)", "job", job.ID, "worker", r.Name)
		return
	}
	if err := r.Store.Complete(lease, err); err != nil {
		slog.Warn("complete rejected", "job", job.ID, "worker", r.Name, "err", err)
	}
}

The store tests drive a fake clock through every lease scenario: retries with backoff, a dead worker's job being reclaimed, the stale worker being fenced out, and recurring jobs skipping missed runs. The runner test uses real time with short durations to check heartbeats and crash recovery end to end.

internal/jobs/jobs_test.go
Go
package jobs
 
import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)
 
type clock struct{ t time.Time }
 
func (c *clock) Now() time.Time          { return c.t }
func (c *clock) Advance(d time.Duration) { c.t = c.t.Add(d) }
 
func newStore() (*Store, *clock) {
	c := &clock{time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)}
	return NewStore(30*time.Second, 10*time.Second, c.Now), c
}
 
func TestClaimOrderAndDelay(t *testing.T) {
	s, c := newStore()
	later := s.Enqueue(Spec{Kind: "a", RunAt: c.Now().Add(time.Minute)})
	first := s.Enqueue(Spec{Kind: "b"})
	l, j, ok := s.Claim("w1")
	if !ok || j.ID != first.ID || j.Attempts != 1 {
		t.Fatalf("claimed %+v", j)
	}
	if _, _, ok := s.Claim("w2"); ok {
		t.Fatal("the delayed job is not due yet")
	}
	s.Complete(l, nil)
	c.Advance(time.Minute)
	if _, j, _ := s.Claim("w2"); j.ID != later.ID {
		t.Errorf("after a minute claimed %q", j.ID)
	}
}
 
func TestRetryBackoffThenDead(t *testing.T) {
	s, c := newStore()
	s.Enqueue(Spec{Kind: "a", MaxAttempts: 3})
	boom := errors.New("upstream 503")
	for attempt, wait := range []time.Duration{10 * time.Second, 20 * time.Second} {
		l, _, ok := s.Claim("w")
		if !ok {
			t.Fatalf("attempt %d not claimable", attempt+1)
		}
		s.Complete(l, boom)
		c.Advance(wait - time.Second)
		if _, _, ok := s.Claim("w"); ok {
			t.Fatalf("claimed during backoff after attempt %d", attempt+1)
		}
		c.Advance(time.Second)
	}
	l, _, _ := s.Claim("w")
	s.Complete(l, boom)
	if j := s.List()[0]; j.State != Dead || j.Attempts != 3 || j.LastError != "upstream 503" {
		t.Errorf("after 3 failures: %+v", j)
	}
	if err := s.Retry(s.List()[0].ID); err != nil {
		t.Fatal(err)
	}
	if _, j, ok := s.Claim("w"); !ok || j.Attempts != 1 {
		t.Errorf("retried job: %+v", j)
	}
}
 
func TestExpiredLeaseIsReclaimedAndStaleWorkerFenced(t *testing.T) {
	s, c := newStore()
	s.Enqueue(Spec{Kind: "a"})
	leaseA, _, _ := s.Claim("A")
	c.Advance(20 * time.Second)
	if err := s.Heartbeat(leaseA); err != nil {
		t.Fatal(err) // lease now runs until 12:00:50
	}
	c.Advance(25 * time.Second)
	if _, _, ok := s.Claim("B"); ok {
		t.Fatal("heartbeat should have kept A's lease alive")
	}
	c.Advance(10 * time.Second) // A is silent: the lease expires
	leaseB, j, ok := s.Claim("B")
	if !ok || j.Attempts != 2 {
		t.Fatalf("B's claim: %v %+v", ok, j)
	}
	if err := s.Complete(leaseA, nil); !errors.Is(err, ErrLeaseLost) {
		t.Errorf("stale complete: %v", err)
	}
	if err := s.Heartbeat(leaseA); !errors.Is(err, ErrLeaseLost) {
		t.Errorf("stale heartbeat: %v", err)
	}
	if err := s.Complete(leaseB, nil); err != nil {
		t.Fatal(err)
	}
	if j := s.List()[0]; j.State != Succeeded || j.Runs != 1 {
		t.Errorf("final: %+v", j)
	}
}
 
func TestLeaseExpiryOnLastAttemptIsDead(t *testing.T) {
	s, c := newStore()
	s.Enqueue(Spec{Kind: "a", MaxAttempts: 1})
	s.Claim("A")
	c.Advance(time.Minute)
	if _, _, ok := s.Claim("B"); ok {
		t.Fatal("no attempts left")
	}
	if j := s.List()[0]; j.State != Dead {
		t.Errorf("state = %s", j.State)
	}
}
 
func TestRecurringSkipsMissedRuns(t *testing.T) {
	s, c := newStore()
	start := c.Now()
	s.Enqueue(Spec{Kind: "report", Every: time.Minute})
	l, _, _ := s.Claim("w")
	s.Complete(l, nil)
	if j := s.List()[0]; j.State != Pending || !j.RunAt.Equal(start.Add(time.Minute)) {
		t.Fatalf("next run: %+v", j)
	}
	c.Advance(3*time.Minute + 30*time.Second) // the system was down for a while
	l, _, _ = s.Claim("w")
	s.Complete(l, nil)
	if j := s.List()[0]; !j.RunAt.Equal(start.Add(4*time.Minute)) || j.Runs != 2 {
		t.Errorf("after downtime: run_at %v, runs %d", j.RunAt.Sub(start), j.Runs)
	}
}
 
func TestPermanentErrorsAreNotRetried(t *testing.T) {
	s, _ := newStore()
	s.Enqueue(Spec{Kind: "a", MaxAttempts: 5})
	l, _, _ := s.Claim("w")
	s.Complete(l, errors.Join(errors.New("invalid email"), ErrPermanent))
	if j := s.List()[0]; j.State != Dead || j.Attempts != 1 {
		t.Errorf("%+v", j)
	}
}
 
func TestRunnerHeartbeatsAndCrashRecovery(t *testing.T) {
	s := NewStore(200*time.Millisecond, 10*time.Millisecond, nil)
	var crashed atomic.Bool
	handlers := map[string]Handler{
		"slow": func(ctx context.Context, j Job) error { // longer than the lease: heartbeats keep it
			select {
			case <-time.After(500 * time.Millisecond):
				return nil
			case <-ctx.Done():
				return ctx.Err()
			}
		},
		"crash": func(ctx context.Context, j Job) error {
			if crashed.CompareAndSwap(false, true) {
				return ErrAbandon
			}
			return nil
		},
	}
	slow := s.Enqueue(Spec{Kind: "slow"})
	crash := s.Enqueue(Spec{Kind: "crash"})
	unknown := s.Enqueue(Spec{Kind: "nope"})
 
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	for _, name := range []string{"w1", "w2"} {
		r := &Runner{Store: s, Name: name, Handlers: handlers, Poll: 10 * time.Millisecond, Heartbeat: 50 * time.Millisecond}
		go r.Run(ctx)
	}
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		done := 0
		for _, j := range s.List() {
			if j.State == Succeeded || j.State == Dead {
				done++
			}
		}
		if done == 3 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	byID := map[string]Job{}
	for _, j := range s.List() {
		byID[j.ID] = j
	}
	if j := byID[slow.ID]; j.State != Succeeded || j.Attempts != 1 {
		t.Errorf("slow job: %+v (heartbeats should prevent a second attempt)", j)
	}
	if j := byID[crash.ID]; j.State != Succeeded || j.Attempts != 2 {
		t.Errorf("crash job: %+v (should be reclaimed after the lease expired)", j)
	}
	if j := byID[unknown.ID]; j.State != Dead {
		t.Errorf("unknown kind: %+v", j)
	}
}

Step 4: the API and demo handlers

Three runners share one store. The handlers are deliberately varied: email succeeds, flaky fails 60% of the time, slow runs for 6 seconds with a 3-second lease (heartbeats keep it alive), and crash kills its worker on the first attempt. A recurring report job runs every 15 seconds.

main.go
Go
package main
 
import (
	"context"
	_ "embed"
	"encoding/json"
	"errors"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"os"
	"time"
 
	"scheduler/internal/jobs"
)
 
//go:embed web/index.html
var indexHTML []byte
 
func sleep(ctx context.Context, d time.Duration) error {
	select {
	case <-time.After(d):
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}
 
var handlers = map[string]jobs.Handler{
	"email": func(ctx context.Context, j jobs.Job) error { return sleep(ctx, 300*time.Millisecond) },
	"flaky": func(ctx context.Context, j jobs.Job) error {
		if err := sleep(ctx, 200*time.Millisecond); err != nil {
			return err
		}
		if rand.Float64() < 0.6 {
			return errors.New("upstream returned 503")
		}
		return nil
	},
	"slow": func(ctx context.Context, j jobs.Job) error { return sleep(ctx, 6*time.Second) },
	"crash": func(ctx context.Context, j jobs.Job) error {
		if j.Attempts == 1 {
			sleep(ctx, 200*time.Millisecond)
			return jobs.ErrAbandon
		}
		return sleep(ctx, 200*time.Millisecond)
	},
	"report": func(ctx context.Context, j jobs.Job) error { return sleep(ctx, 500*time.Millisecond) },
}
 
func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(v)
}
 
func newServer(s *jobs.Store, now func() time.Time) http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		w.Write(indexHTML)
	})
	mux.HandleFunc("GET /api/jobs", func(w http.ResponseWriter, r *http.Request) {
		writeJSON(w, http.StatusOK, map[string]any{"now": now(), "jobs": s.List()})
	})
	mux.HandleFunc("POST /api/jobs", func(w http.ResponseWriter, r *http.Request) {
		var in struct {
			Kind         string  `json:"kind"`
			Payload      string  `json:"payload"`
			DelaySeconds float64 `json:"delay_seconds"`
			EverySeconds float64 `json:"every_seconds"`
			MaxAttempts  int     `json:"max_attempts"`
		}
		if json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<14)).Decode(&in) != nil {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
			return
		}
		if _, ok := handlers[in.Kind]; !ok {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown kind"})
			return
		}
		if in.DelaySeconds < 0 || in.EverySeconds < 0 || in.MaxAttempts < 0 || in.MaxAttempts > 20 {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "out of range"})
			return
		}
		j := s.Enqueue(jobs.Spec{
			Kind: in.Kind, Payload: in.Payload, MaxAttempts: in.MaxAttempts,
			RunAt: now().Add(time.Duration(in.DelaySeconds * float64(time.Second))),
			Every: time.Duration(in.EverySeconds * float64(time.Second)),
		})
		writeJSON(w, http.StatusCreated, j)
	})
	mux.HandleFunc("POST /api/jobs/{id}/retry", func(w http.ResponseWriter, r *http.Request) {
		if err := s.Retry(r.PathValue("id")); err != nil {
			writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()})
			return
		}
		w.WriteHeader(http.StatusNoContent)
	})
	return mux
}
 
func main() {
	s := jobs.NewStore(3*time.Second, time.Second, nil)
	s.Enqueue(jobs.Spec{Kind: "report", Every: 15 * time.Second})
	for _, name := range []string{"worker-1", "worker-2", "worker-3"} {
		r := &jobs.Runner{Store: s, Name: name, Handlers: handlers, Poll: 200 * time.Millisecond, Heartbeat: time.Second}
		go r.Run(context.Background())
	}
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}
	slog.Info("listening", "port", port)
	srv := &http.Server{Addr: ":" + port, Handler: newServer(s, time.Now), ReadHeaderTimeout: 5 * time.Second}
	if err := srv.ListenAndServe(); err != nil {
		slog.Error("server", "err", err)
		os.Exit(1)
	}
}
main_test.go
Go
package main
 
import (
	"encoding/json"
	"net/http/httptest"
	"strings"
	"testing"
	"time"
 
	"scheduler/internal/jobs"
)
 
func TestAPI(t *testing.T) {
	now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	clock := func() time.Time { return now }
	s := jobs.NewStore(time.Second, time.Second, clock)
	h := newServer(s, clock)
	call := func(method, url, body string) int {
		rec := httptest.NewRecorder()
		h.ServeHTTP(rec, httptest.NewRequest(method, url, strings.NewReader(body)))
		return rec.Code
	}
	if code := call("POST", "/api/jobs", `{"kind":"email","delay_seconds":5}`); code != 201 {
		t.Fatalf("enqueue: %d", code)
	}
	for _, bad := range []string{`{"kind":"nope"}`, `{"kind":"email","delay_seconds":-1}`, `{`} {
		if code := call("POST", "/api/jobs", bad); code != 400 {
			t.Errorf("%s: %d", bad, code)
		}
	}
	if code := call("POST", "/api/jobs/job-0001/retry", ""); code != 409 {
		t.Errorf("retrying a pending job: %d", code)
	}
 
	rec := httptest.NewRecorder()
	h.ServeHTTP(rec, httptest.NewRequest("GET", "/api/jobs", nil))
	var out struct{ Jobs []jobs.Job }
	json.NewDecoder(rec.Body).Decode(&out)
	if len(out.Jobs) != 1 || out.Jobs[0].State != jobs.Pending || !out.Jobs[0].RunAt.Equal(now.Add(5*time.Second)) {
		t.Errorf("jobs = %+v", out.Jobs)
	}
}

Step 5: the dashboard

web/index.html
HTML
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Job scheduler</title>
  <style>
    body { font-family: system-ui, sans-serif; max-width: 980px; margin: 2rem auto; padding: 0 1rem; }
    button { padding: .4rem .7rem; border-radius: 8px; border: 1px solid #999; background: #fff; cursor: pointer; font: inherit; margin: 0 .3rem .4rem 0; }
    table { width: 100%; border-collapse: collapse; font-size: 14px; margin-top: 1rem; }
    th, td { text-align: left; padding: .4rem .5rem; border-bottom: 1px solid #ddd; }
    td.mono { font-family: monospace; }
    .st { font: 12px monospace; padding: .1rem .4rem; border-radius: 4px; color: #fff; }
    .pending { background: #b57614; } .running { background: #076678; } .succeeded { background: #79740e; } .dead { background: #9d0006; }
  </style>
</head>
<body>
  <h1>Job scheduler</h1>
  <p>Three workers, 3 s lease, heartbeat every 1 s, retry backoff 1 s doubling.</p>
  <div id="buttons"></div>
  <table>
    <thead><tr><th>id</th><th>kind</th><th>state</th><th>attempt</th><th>runs</th><th>worker</th><th>next / lease</th><th>last error</th><th></th></tr></thead>
    <tbody id="rows"></tbody>
  </table>
  <script>
    const presets = [
      ["email", { kind: "email" }],
      ["email in 5 s", { kind: "email", delay_seconds: 5 }],
      ["flaky (60% fail)", { kind: "flaky", max_attempts: 4 }],
      ["slow (6 s)", { kind: "slow" }],
      ["crash worker", { kind: "crash" }],
      ["every 10 s", { kind: "email", every_seconds: 10 }],
    ];
    for (const [label, body] of presets) {
      const b = document.createElement("button");
      b.textContent = "+ " + label;
      b.onclick = () => fetch("/api/jobs", { method: "POST", body: JSON.stringify(body) }).then(refresh);
      buttons.append(b);
    }
    const secs = (t, now) => { const d = (new Date(t) - now) / 1000; return (d >= 0 ? "in " : "") + Math.abs(d).toFixed(1) + " s" + (d < 0 ? " ago" : ""); };
    async function refresh() {
      const { now, jobs } = await (await fetch("/api/jobs")).json();
      const n = new Date(now);
      rows.innerHTML = "";
      for (const j of jobs.reverse()) {
        const tr = document.createElement("tr");
        const when = j.state === "running" ? "lease " + secs(j.lease_until, n) : j.state === "pending" ? secs(j.run_at, n) : "";
        const cells = [j.id, j.kind + (j.every ? ` (every ${j.every / 1e9} s)` : ""), null, `${j.attempts}/${j.max_attempts}`, j.runs, j.worker || "", when, j.last_error || ""];
        cells.forEach((v, i) => {
          const td = document.createElement("td");
          if (v === null) {
            const s = document.createElement("span");
            s.className = "st " + j.state;
            s.textContent = j.state;
            td.append(s);
          } else td.textContent = v;
          if (i === 0 || i === 6) td.className = "mono";
          tr.append(td);
        });
        const td = document.createElement("td");
        if (j.state === "dead") {
          const b = document.createElement("button");
          b.textContent = "retry";
          b.onclick = () => fetch(`/api/jobs/${j.id}/retry`, { method: "POST" }).then(refresh);
          td.append(b);
        }
        tr.append(td);
        rows.append(tr);
      }
    }
    setInterval(refresh, 500);
    refresh();
  </script>
</body>
</html>

Step 6: run it

Shell
go test -race ./...
go run .
# open http://localhost:8080
curl -s -X POST localhost:8080/api/jobs -d '{"kind":"crash"}'
# watch: running on one worker, then silence; ~3 s later another worker claims it (attempt 2) and it succeeds

Things to try on the dashboard: add several flaky jobs and watch backoff spread out the retries, with some ending dead (press retry). Add a slow job: its lease keeps moving forward while it runs. Add crash worker: the log prints the simulated crash, the lease counts down past zero, and another worker takes over.

Step 7: toward production

  • Persist jobs in Postgres with the SKIP LOCKED query from section 4, and put every lease check in the WHERE clause of the UPDATE (WHERE id = $1 AND worker = $2 AND attempt = $3).
  • Graceful shutdown: on SIGTERM, stop claiming, let running jobs finish within a deadline, and let leases expire for anything left.
  • Unique jobs: a unique key column (report:2026-01-01) so concurrent producers cannot enqueue the same scheduled run twice.
  • Metrics: queue depth by kind, age of the oldest due job (the best single health signal), attempts and dead jobs.

Summary#

  • A lease (owner + attempt + expiry) makes crash recovery automatic: an expired lease means anyone may take the job.
  • Heartbeats keep long jobs alive; the lease must comfortably exceed the heartbeat interval.
  • Fence every write with the lease so a worker that paused cannot overwrite its successor's result.
  • Jobs run at least once, so make them idempotent; retry with backoff, cap attempts, and keep dead jobs visible.
  • Schedule recurring runs from the schedule, not from "now", and skip runs missed during downtime.

Check your understanding

Answer before scrolling back up. If you miss one, reread the matching section, then try again.

  1. Q1.A worker takes a job and then its machine dies. How does the job get done?

  2. Q2.Why must long jobs send heartbeats?

  3. Q3.Worker A paused for a long GC; its lease expired and worker B took the job. A wakes up and reports success. What should happen?

  4. Q4.What does SELECT ... FOR UPDATE SKIP LOCKED give a Postgres-backed job queue?

  5. Q5.Since a job may run more than once (at-least-once), what should job code be?

Practice in Go

Write a table test first, then the solution. On LeetCode, choose Go as the language. Open a hint only after 15 minutes stuck.

  • Task Scheduler ↗medium
    Show hint

    Cooldowns between equal tasks are the same idea as backoff before a retry.

  • Single-Threaded CPU ↗medium
    Show hint

    A heap of available tasks ordered by time and index: the claim order of a scheduler.

  • Cron expressionsbuild

    Support recurring jobs written as cron expressions ("*/5 * * * *", "0 9 * * 1-5") instead of a fixed interval. Write next(expr, after time.Time) with table-driven tests, including month ends.

  • Postgres storebuild

    Implement the store on Postgres: a jobs table, Claim with UPDATE ... WHERE id = (SELECT id ... FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *, and lease checks in WHERE clauses. Run three workers and show no job runs twice concurrently.

Finished this lesson?

Mark it done to track your progress. It is saved in this browser only.