41. Capstone: a distributed-style job queue with persistence

Goal: combine channels, workers, context, encoding/json, and file persistence into a small but real job queue — the kind of tool that sits behind "process this batch of uploaded files" or "send this batch of emails" in a real backend.

// queue/job.go
package queue

type Status string

const (
	Pending   Status = "pending"
	Running   Status = "running"
	Succeeded Status = "succeeded"
	Failed    Status = "failed"
)

type Job struct {
	ID      string `json:"id"`
	Payload string `json:"payload"`
	Status  Status `json:"status"`
	Error   string `json:"error,omitempty"`
}

// queue/queue.go
package queue

import (
	"context"
	"encoding/json"
	"os"
	"sync"
)

type Queue struct {
	mu       sync.Mutex
	jobs     map[string]*Job
	pending  chan *Job
	statePath string
}

func New(statePath string, bufferSize int) *Queue {
	q := &Queue{
		jobs:      map[string]*Job{},
		pending:   make(chan *Job, bufferSize),
		statePath: statePath,
	}
	_ = q.load() // teaching example: production code should return/log this error
	return q
}

func (q *Queue) Enqueue(id, payload string) *Job {
	q.mu.Lock()
	defer q.mu.Unlock()
	j := &Job{ID: id, Payload: payload, Status: Pending}
	q.jobs[id] = j
	q.pending <- j
	_ = q.persistLocked()
	return j
}

func (q *Queue) Get(id string) (*Job, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	j, ok := q.jobs[id]
	return j, ok
}

func (q *Queue) Run(ctx context.Context, workers int, handler func(string) error) {
	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case j, ok := <-q.pending:
					if !ok {
						return
					}
					q.process(j, handler)
				}
			}
		}()
	}
	wg.Wait()
}

func (q *Queue) process(j *Job, handler func(string) error) {
	q.setStatus(j, Running, "")
	if err := handler(j.Payload); err != nil {
		q.setStatus(j, Failed, err.Error())
		return
	}
	q.setStatus(j, Succeeded, "")
}

func (q *Queue) setStatus(j *Job, status Status, errMsg string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	j.Status = status
	j.Error = errMsg
	_ = q.persistLocked()
}

// caller must hold q.mu
func (q *Queue) persistLocked() error {
	f, err := os.Create(q.statePath)
	if err != nil {
		return err
	}
	if err := json.NewEncoder(f).Encode(q.jobs); err != nil {
		_ = f.Close()
		return err
	}
	return f.Close()
}

func (q *Queue) load() error {
	f, err := os.Open(q.statePath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil // no state file yet — fresh start
		}
		return err
	}
	defer f.Close()
	if err := json.NewDecoder(f).Decode(&q.jobs); err != nil {
		return err
	}
	for _, j := range q.jobs {
		if j.Status == Pending || j.Status == Running {
			j.Status = Pending
			q.pending <- j // re-enqueue anything that didn't finish last run
		}
	}
	return nil
}

// cmd/jobqueue/main.go
package main

import (
	"context"
	"fmt"
	"log"
	"os/signal"
	"syscall"
	"time"

	"example.com/jobqueue/queue"
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	q := queue.New("jobs.json", 100)

	for i := 0; i < 5; i++ {
		q.Enqueue(fmt.Sprintf("job-%d", i), fmt.Sprintf("payload-%d", i))
	}

	q.Run(ctx, 3, func(payload string) error {
		log.Println("processing", payload)
		time.Sleep(500 * time.Millisecond)
		return nil
	})
}

This gives you: a buffered channel as the in-memory work queue, persistence to disk on every state change (so a crash doesn't silently lose which jobs were pending), and signal.NotifyContext (Go 1.16+) to turn Ctrl-C into a clean, cancellable shutdown instead of an abrupt kill. Scaling this to actually distributed (multiple processes, not just multiple goroutines) means swapping the in-memory channel for a shared backend — Redis, PostgreSQL with SELECT ... FOR UPDATE SKIP LOCKED, or a message broker — while keeping the exact same Job/Status model.