JobQueue — Distributed Background Job Processor in Go

A production-grade background job processing system built with Go and MongoDB. Supports distributed workers, priority queuing, retry logic, scheduled jobs, and structured observability.

---

Table of Contents

• Overview
• Architecture
• Project Structure
• Job Lifecycle
• Getting Started
• API Reference
• Configuration
• Worker System
• Retry & Backoff
• Observability
• Design Decisions

---

Overview

Feature	Details
Language	Go 1.21+
Database	MongoDB
HTTP Framework	Gin
Job Types	email, webhook, report
Priority Levels	1 (lowest) → 5 (highest)
Concurrency	Configurable worker pool (goroutines)
Locking	Atomic findOneAndUpdate — no external lock manager needed
Retry Strategy	Exponential backoff with jitter
Scheduling	Delayed jobs via run_at field

---

Architecture

┌─────────────────────────────────────────────────────────┐
│                        HTTP API (Gin)                    │
│         POST /jobs   GET /jobs/:id   DELETE /jobs/:id    │
└──────────────────────────┬──────────────────────────────┘
                           │
                    ┌──────▼──────┐
                    │  Use Cases  │  (business logic)
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │  Repository │  (MongoDB)
                    └──────┬──────┘
                           │
              ┌────────────▼────────────┐
              │         MongoDB          │
              │   jobs collection        │
              │   index: status +        │
              │   priority + run_at      │
              └────────────▲────────────┘
                           │
              ┌────────────┴────────────┐
              │      Worker Pool         │
              │  ┌────┐ ┌────┐ ┌────┐  │
              │  │ W1 │ │ W2 │ │ W3 │  │  ← goroutines
              │  └────┘ └────┘ └────┘  │
              │   poll → lock → exec    │
              └─────────────────────────┘

---

Project Structure

jobqueue/
├── cmd/
│   └── server/
│       └── main.go              # Entry point — wires everything together
├── internal/
│   ├── config/
│   │   └── config.go            # Env-based configuration
│   ├── domain/
│   │   └── job.go               # Job struct, statuses, constants
│   ├── handler/
│   │   └── job_handler.go       # HTTP handlers (Gin)
│   ├── usecase/
│   │   └── job_usecase.go       # Business logic
│   ├── infrastructure/
│   │   ├── mongodb/
│   │   │   └── job_repository.go  # DB queries, atomic locking
│   │   └── worker/
│   │       ├── worker.go          # Worker goroutine + polling loop
│   │       ├── pool.go            # Worker pool management
│   │       └── handlers.go        # Job type executors (email, webhook, report)
│   ├── logger/
│   │   └── logger.go            # Structured slog wrapper
│   └── middleware/
│       ├── timeout.go           # Request + DB + job timeouts
│       └── wiring.go            # Usage reference / docs
├── docker-compose.yml
├── Dockerfile
├── Makefile
├── .env.example
├── .gitignore
└── README.md

---

Job Lifecycle

POST /jobs
    │
    ▼
status: "pending"
run_at: now (or future for scheduled)
    │
    ▼  (worker polls)
status: "running"
locked_by: "worker-3"
locked_at: <now>
    │
    ├── success ──► status: "completed"
    │
    └── failure
            │
            ├── attempts < max_attempts
            │       │
            │       ▼
            │   status: "pending"
            │   run_at: now + backoff   (exponential)
            │   attempts: +1
            │
            └── attempts >= max_attempts
                        │
                        ▼
                    status: "failed"
                    last_error: "..."

---

Getting Started

Prerequisites

• Docker + Docker Compose

That's it. No Go installation needed to run.

Run with Docker

# 1. Clone the repo
git clone https://github.com/yourname/jobqueue.git
cd jobqueue

# 2. Copy environment config
cp .env.example .env

# 3. Start everything (Go app + MongoDB)
make up

# 4. Watch logs
make logs

The API will be available at http://localhost:8080.

Run locally (with Go installed)

# Requires Go 1.21+ and a running MongoDB instance
cp .env.example .env
# Edit .env — set MONGO_URI to your MongoDB connection string

go mod download
go run cmd/server/main.go

Makefile Commands

make up          # docker compose up --build -d
make down        # docker compose down
make logs        # follow all container logs
make simulate    # seed test jobs and watch workers process them
make restart     # down + up
make clean       # remove containers, volumes, images

---

API Reference

Create a Job

POST /jobs
Content-Type: application/json

{
  "type": "email",
  "payload": {
    "to": "user@example.com",
    "subject": "Welcome!"
  },
  "priority": 3,
  "max_attempts": 5,
  "run_at": "2025-01-15T10:00:00Z"
}

Omit run_at to run immediately. priority defaults to 1 if omitted.

Response 201

{
  "id": "64f1a2b3c4d5e6f7a8b9c0d1",
  "status": "pending"
}

---

Get Job Status

GET /jobs/:id

Response 200

{
  "_id": "64f1a2b3c4d5e6f7a8b9c0d1",
  "type": "email",
  "status": "completed",
  "priority": 3,
  "attempts": 1,
  "max_attempts": 5,
  "created_at": "2025-01-15T09:55:00Z",
  "updated_at": "2025-01-15T10:00:03Z"
}

---

List Jobs

GET /jobs?status=failed&type=email&priority=3&page=1&limit=20

All query params are optional. Default: page=1, limit=20.

---

Cancel a Job

DELETE /jobs/:id

Only pending jobs can be cancelled. Returns 409 if the job is already running or completed.

---

Configuration

Copy .env.example to .env and adjust:

# Server
PORT=8080
ENV=development          # development | production

# MongoDB
MONGO_URI=mongodb://localhost:27017
MONGO_DB=jobqueue

# Worker pool
WORKER_COUNT=5           # number of concurrent worker goroutines
POLL_INTERVAL=2s         # how often workers check for new jobs
LOCK_TIMEOUT=5m          # jobs locked longer than this are re-queued

# Timeouts
REQUEST_TIMEOUT=10s      # max HTTP request duration
DB_TIMEOUT=5s            # max MongoDB operation duration
JOB_TIMEOUT=60s          # max single job execution duration

---

Worker System

Workers run as goroutines in a pool. Each worker independently polls MongoDB on a configurable interval.

Atomic Locking

The critical section — claiming a job — uses a single atomic findOneAndUpdate:

// Pseudo-code for what the worker does
db.jobs.findOneAndUpdate(
  {
    status: "pending",
    run_at: { $lte: now },
    $or: [
      { locked_by: null },
      { locked_at: { $lt: now - lockTimeout } }   // stale lock recovery
    ]
  },
  {
    $set: {
      status: "running",
      locked_by: "worker-3",
      locked_at: now
    }
  },
  { sort: { priority: -1, run_at: 1 } }   // priority DESC, run_at ASC
)

If two workers poll simultaneously, MongoDB's atomic update guarantees exactly one wins. The other gets null back and moves on.

Stale Lock Recovery

If a worker crashes mid-job, the lock is never released. The $or condition above automatically makes stale jobs (locked longer than LOCK_TIMEOUT) available again — no separate cleanup job needed.

Graceful Shutdown

On SIGINT / SIGTERM:

1. Worker pool stops accepting new jobs
2. In-flight jobs are allowed to finish (up to JOB_TIMEOUT)
3. MongoDB connections are closed cleanly

---

Retry & Backoff

On failure, if attempts < max_attempts the job is rescheduled with exponential backoff:

attempt 1 → retry in  2s  + jitter
attempt 2 → retry in  4s  + jitter
attempt 3 → retry in  8s  + jitter
attempt 4 → retry in 16s  + jitter
attempt 5 → mark as FAILED

The jitter (±20%) prevents thundering herd — if 100 jobs fail simultaneously, they don't all retry at the exact same instant.

---

Observability

Structured Logging

Every log line is a JSON object (production) or human-readable text (development):

{
  "time": "2025-01-15T10:00:01Z",
  "level": "INFO",
  "msg": "job completed",
  "job_id": "64f1a2b3c4d5e6f7a8b9c0d1",
  "job_type": "email",
  "worker_id": "worker-3",
  "attempt": 1,
  "duration_ms": 243,
  "source": "worker/worker.go:87"
}

Fields attached at job start propagate to all subsequent log lines for that job automatically via logger.WithFields.

Key Log Events

Event	Level	When
job created	INFO	POST /jobs succeeds
job locked	DEBUG	Worker claims a job
job started	INFO	Execution begins
job completed	INFO	Execution succeeds
job failed, retrying	WARN	Failure, attempts remaining
job failed permanently	ERROR	Attempts exhausted
stale lock recovered	WARN	Dead worker's job reclaimed
worker shutting down	INFO	SIGINT received

---

Design Decisions

Why MongoDB instead of Redis/Postgres for the queue? MongoDB's atomic findOneAndUpdate is sufficient for a job queue without needing a separate lock manager. It also stores payloads + history in the same document, keeping the stack simple.

Why not use an existing queue (BullMQ, RabbitMQ)? This system intentionally implements the queue mechanics from scratch to make the distributed locking, retry, and scheduling logic transparent and educational.

Why log/slog over zap or zerolog? slog is part of the Go standard library since 1.21. No external dependency, same structured logging API, good enough performance for this scale.

Why Gin over standard net/http? Routing, middleware chaining, and context handling with minimal boilerplate. For a pure microservice net/http + chi would also work.