Build a Priority Queue Scheduler in Go

Build a heap-based job scheduler API with tenant fairness in pure Go.

Introduction

30 Second Summary

When multiple teams submit thousands of test jobs to a shared cluster, someone has to decide what runs first and who gets a fair share. That decision-maker is a scheduler.

In this project, you will build a control plane server in Go that accepts test job submissions, routes them through a priority queue with tenant fairness, and shuts down gracefully on termination signals.

What You'll Build

A production-patterned HTTP API that manages job scheduling using a heap-based priority queue with weighted fair queueing and a ticker-based scheduler loop.

By the end of this project, you'll have:

  • A heap-based priority queue implementing Go's container/heap interface with per-tenant weight tracking.
  • An HTTP API with 4 endpoints using Go 1.22+ method-based routing and zero external dependencies.
  • A scheduler goroutine running a ticker-based loop with graceful shutdown via OS signal handling.
  • Secret Mission: implement admission control with a per-tenant token bucket rate limiter.

Are there any prerequisites?

You should have basic Go knowledge including functions, structs, interfaces, and goroutines. You should also have Go and VS Code already installed on your machine.

This is Project 1 of a 3-part series. By the end, you will have the "brain" of a distributed system that decides what runs, when, and for whom.

Set Up the Project Scaffold

A distributed test scheduler has many moving parts: a priority queue, an HTTP API, a scheduler loop, and shared data models. Keeping these in separate packages from the start prevents a tangled codebase as the system grows.

In this step, you will initialize a Go module, create the package directory layout, and define the core Job model that every other package will depend on.

In this step, get ready to:
  • Initialize a Go module and create the package directory layout.
  • Define the Job model and state machine.
  • Verify the scaffold compiles with zero errors.
Initialize the Go module and create the directory layout

Go modules need a root directory with a go.mod file that declares the module path. The directory layout separates concerns into packages that compile independently.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir testscheduler
cd testscheduler
  • Initialize the Go module by running:
go mod init github.com/yourname/testscheduler

What does this command do?

go mod init creates a go.mod file that declares your module's import path.

Every other package in this project will use github.com/yourname/testscheduler/internal/... as its import prefix.

You should see output similar to:

go: creating new go.mod: module github.com/yourname/testscheduler
  • Create the package directory layout by running:
mkdir -p cmd/scheduler internal/queue internal/scheduler internal/api internal/model

Why this directory layout?

  • cmd/scheduler/ holds the executable entry point (the main package).
  • internal/ packages are only importable within this module. Go enforces this at compile time.
  • Each subdirectory under internal/ maps to one responsibility: queue logic, scheduling, HTTP API, and shared models.
  • Confirm the directories exist by running:
ls cmd/scheduler internal/queue internal/scheduler internal/api internal/model

Each directory should appear in the output with no errors.

Seeing 'No such file or directory'?

  • Make sure you ran mkdir -p with the -p flag. Without it, nested directories like cmd/scheduler will fail if the parent does not exist yet.
  • Confirm you are inside the testscheduler folder by running pwd.

help me fix my directory layout

  • Open the testscheduler folder in VS Code by running:
code .
Define the job model and state machine

Every component in this scheduler needs a shared definition of what a "job" is and which states it can move through. The Job struct and its state machine live in the model package so the queue, API, and scheduler can all import it.

What is a state machine?

A state machine defines the valid states an entity can be in and which transitions between states are allowed.

For our jobs, a Pending job can move to Queued but never directly to Completed. This prevents impossible states and makes bugs easier to catch.

  • In VS Code, right-click the internal/model folder in the Explorer sidebar.
  • Select New File and name it job.go.

You will build this file in three parts.

  • Add the package declaration, imports, and state type definitions by pasting this code into job.go:
package model

import (
	"fmt"
	"time"
)

// JobState represents the lifecycle state of a job
type JobState string

// Valid job states that form the state machine
const (
	StatePending   JobState = "pending"
	StateQueued    JobState = "queued"
	StateScheduled JobState = "scheduled"
	StateRunning   JobState = "running"
	StateCompleted JobState = "completed"
	StateFailed    JobState = "failed"
)

What does this code do?

  • JobState is a custom string type. Using a named type instead of raw strings means the compiler catches typos like "pendng" at build time.
  • The six constants define every valid state a job can occupy during its lifecycle.
  • Add the Job struct below the constants in the same file:
// Job represents a test job submitted to the scheduler
type Job struct {
	ID          string        `json:"id"`
	Tenant      string        `json:"tenant"`
	TestSuite   string        `json:"test_suite"`
	Priority    int           `json:"priority"`
	State       JobState      `json:"state"`
	Delay       time.Duration `json:"delay,omitempty"`
	SubmittedAt time.Time     `json:"submitted_at"`
	ScheduledAt time.Time     `json:"scheduled_at,omitempty"`
	CompletedAt time.Time     `json:"completed_at,omitempty"`
	Index       int           `json:"-"`
}

What does this struct represent?

  • Each field maps to something the scheduler needs to make decisions: Tenant identifies who submitted the job, Priority determines queue ordering.
  • Index is tagged json:"-" because it is an internal field used by the heap. It should never appear in API responses.
  • Delay lets callers submit jobs that should not be queued immediately.
  • Add the state transition logic below the Job struct:
// validTransitions defines which state changes are allowed
var validTransitions = map[JobState][]JobState{
	StatePending:   {StateQueued, StateFailed},
	StateQueued:    {StateScheduled, StateFailed},
	StateScheduled: {StateRunning, StateFailed},
	StateRunning:   {StateCompleted, StateFailed},
}

// TransitionTo moves the job to the next state if the transition is valid
func (j *Job) TransitionTo(next JobState) error {
	allowed, ok := validTransitions[j.State]
	if !ok {
		return fmt.Errorf("no transitions from state %s", j.State)
	}
	for _, s := range allowed {
		if s == next {
			j.State = next
			return nil
		}
	}
	return fmt.Errorf("invalid transition: %s -> %s", j.State, next)
}

What does TransitionTo do?

  • The validTransitions map acts as a whitelist. Each state maps to the states it is allowed to move to.
  • TransitionTo checks the whitelist and returns an error if the requested transition is not permitted.
  • Every state can fail, so StateFailed appears in every transition list.
  • Save internal/model/job.go.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Here is the complete internal/model/job.go file:

package model

import (
	"fmt"
	"time"
)

type JobState string

const (
	StatePending   JobState = "pending"
	StateQueued    JobState = "queued"
	StateScheduled JobState = "scheduled"
	StateRunning   JobState = "running"
	StateCompleted JobState = "completed"
	StateFailed    JobState = "failed"
)

type Job struct {
	ID          string        `json:"id"`
	Tenant      string        `json:"tenant"`
	TestSuite   string        `json:"test_suite"`
	Priority    int           `json:"priority"`
	State       JobState      `json:"state"`
	Delay       time.Duration `json:"delay,omitempty"`
	SubmittedAt time.Time     `json:"submitted_at"`
	ScheduledAt time.Time     `json:"scheduled_at,omitempty"`
	CompletedAt time.Time     `json:"completed_at,omitempty"`
	Index       int           `json:"-"`
}

var validTransitions = map[JobState][]JobState{
	StatePending:   {StateQueued, StateFailed},
	StateQueued:    {StateScheduled, StateFailed},
	StateScheduled: {StateRunning, StateFailed},
	StateRunning:   {StateCompleted, StateFailed},
}

func (j *Job) TransitionTo(next JobState) error {
	allowed, ok := validTransitions[j.State]
	if !ok {
		return fmt.Errorf("no transitions from state %s", j.State)
	}
	for _, s := range allowed {
		if s == next {
			j.State = next
			return nil
		}
	}
	return fmt.Errorf("invalid transition: %s -> %s", j.State, next)
}
Verify the scaffold compiles

Before building any logic on top of this scaffold, you need to confirm that Go can compile everything without errors. A broken import path or misplaced file now would cascade into confusing errors later.

  • Create the entry point file. In VS Code, right-click the cmd/scheduler folder in the Explorer sidebar.
  • Select New File and name it main.go.
  • Paste this placeholder code into main.go:
package main

func main() {}

Why is main empty?

This placeholder satisfies Go's requirement that a cmd/ package must have a main function to compile. You will wire up the real server, scheduler, and signal handling in a later step.

  • Save cmd/scheduler/main.go.
  • Return to your terminal and run the build check:
go build ./...

If the build succeeds, you will see no output at all. The command returns silently on success.

Seeing a compile error?

  • If you see cannot find package, confirm you ran go mod init from inside the testscheduler folder. A go.mod file should exist in that folder.
  • If you see a syntax error in job.go, check that the import block includes both "fmt" and "time". Missing imports cause undefined-type errors.
  • Make sure main.go is inside cmd/scheduler/ and not in the project root. Go expects package main files to live in their own directory.

help me fix my Go build error

Your scaffold is locked in. You have a clean module, an organized directory layout, and a compiled Job model with a state machine that will guard every state transition in the system. Next up, you will build the heap-based priority queue that decides which job runs first.

Build the Heap-Based Priority Queue

Your project scaffold compiles and the job model defines every state a job can pass through. Now you need the engine that decides which job runs next.

A priority queue answers that question by always surfacing the highest-priority item first. You will implement one using Go's container/heap interface, then wrap it in a thread-safe struct that tracks per-tenant usage for fairness.

In this step, get ready to:
  • Implement Go's heap.Interface on a custom JobHeap type for priority ordering.
  • Build a mutex-protected SchedulerQueue with enqueue, dequeue, and query methods.
  • Add weighted fair queueing so no single tenant starves others.
Implement the priority queue

Go's container/heap package provides heap operations, but it needs you to supply the ordering logic. You do this by defining a type that satisfies five methods: Len, Less, Swap, Push, and Pop.

  • Create a new file at internal/queue/priority_queue.go and add the following code:
package queue

import (
	"container/heap"

	"github.com/yourname/testscheduler/internal/model"
)

// JobHeap is a min-heap of jobs ordered by priority
type JobHeap []*model.Job

// Len returns the number of jobs in the heap
func (h JobHeap) Len() int { return len(h) }

// Less determines ordering: lower priority number = higher urgency
// Ties broken by submission time (earlier submitted = dequeued first)
func (h JobHeap) Less(i, j int) bool {
	if h[i].Priority == h[j].Priority {
		return h[i].SubmittedAt.Before(h[j].SubmittedAt)
	}
	return h[i].Priority < h[j].Priority
}

// Swap exchanges two jobs and updates their index fields
func (h JobHeap) Swap(i, j int) {
	h[i], h[j] = h[j], h[i]
	h[i].Index = i
	h[j].Index = j
}

// Push adds a job to the heap (called by heap.Push)
func (h *JobHeap) Push(x any) {
	n := len(*h)
	job := x.(*model.Job)
	job.Index = n
	*h = append(*h, job)
}

// Pop removes and returns the highest-priority job (called by heap.Pop)
func (h *JobHeap) Pop() any {
	old := *h
	n := len(old)
	job := old[n-1]
	old[n-1] = nil
	job.Index = -1
	*h = old[:n-1]
	return job
}

// Compile-time check: JobHeap must satisfy heap.Interface
var _ heap.Interface = (*JobHeap)(nil)

What does this code do?

This file defines a JobHeap type (a slice of Job pointers) that implements Go's heap.Interface.

  • Less compares by Priority (lower number = higher urgency). When two jobs share the same priority, submission time breaks the tie so the earlier job wins.
  • Swap updates each job's Index field. This lets the heap call heap.Fix to re-sort a job in O(log n) if its priority changes later.
  • Pop nils the removed slot to avoid memory leaks, then sets Index = -1 so you know the job is no longer in the heap.
  • Save internal/queue/priority_queue.go.
  • Verify the code compiles by running:
go build ./...

You should see zero output, meaning no errors.

Seeing a compile error?

  • Check that the import path matches your module name exactly: github.com/yourname/testscheduler/internal/model.
  • Make sure internal/model/job.go exists and its package declaration is package model.
  • Confirm all five methods use the exact receiver types shown: JobHeap (value receiver) for Len/Less/Swap and *JobHeap (pointer receiver) for Push/Pop.

Help me debug my priority_queue.go compile error

✔️ Awesome, I've got everything!

Great. Make sure you have saved priority_queue.go before moving on.

ⓧ I'd like to double check the full code

package queue

import (
	"container/heap"

	"github.com/yourname/testscheduler/internal/model"
)

type JobHeap []*model.Job

func (h JobHeap) Len() int { return len(h) }

func (h JobHeap) Less(i, j int) bool {
	if h[i].Priority == h[j].Priority {
		return h[i].SubmittedAt.Before(h[j].SubmittedAt)
	}
	return h[i].Priority < h[j].Priority
}

func (h JobHeap) Swap(i, j int) {
	h[i], h[j] = h[j], h[i]
	h[i].Index = i
	h[j].Index = j
}

func (h *JobHeap) Push(x any) {
	n := len(*h)
	job := x.(*model.Job)
	job.Index = n
	*h = append(*h, job)
}

func (h *JobHeap) Pop() any {
	old := *h
	n := len(old)
	job := old[n-1]
	old[n-1] = nil
	job.Index = -1
	*h = old[:n-1]
	return job
}

var _ heap.Interface = (*JobHeap)(nil)
Build the thread-safe queue wrapper

The raw heap is not safe for concurrent access. In a real server, multiple HTTP handlers and the scheduler goroutine will touch the queue simultaneously. You need a wrapper that locks the heap behind a sync.Mutex.

  • Create a new file at internal/queue/scheduler_queue.go and add the struct definition, constructor, and core queue operations:
package queue

import (
	"container/heap"
	"sync"

	"github.com/yourname/testscheduler/internal/model"
)

// SchedulerQueue wraps JobHeap with mutex protection and tenant tracking
type SchedulerQueue struct {
	mu            sync.Mutex
	heap          JobHeap
	jobs          map[string]*model.Job
	tenantCounts  map[string]int
	tenantWeights map[string]int
	pending       []*model.Job
}

// NewSchedulerQueue creates an initialized queue ready for use
func NewSchedulerQueue() *SchedulerQueue {
	sq := &SchedulerQueue{
		jobs:          make(map[string]*model.Job),
		tenantCounts:  make(map[string]int),
		tenantWeights: make(map[string]int),
	}
	heap.Init(&sq.heap)
	return sq
}

// Enqueue adds a job to the priority heap with fairness-adjusted priority
func (sq *SchedulerQueue) Enqueue(job *model.Job) {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	job.State = model.StateQueued
	job.Priority = sq.effectivePriority(job)
	sq.jobs[job.ID] = job
	sq.tenantCounts[job.Tenant]++
	heap.Push(&sq.heap, job)
}

// AddPending stores a delayed job that will be promoted later
func (sq *SchedulerQueue) AddPending(job *model.Job) {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	job.State = model.StatePending
	sq.jobs[job.ID] = job
	sq.pending = append(sq.pending, job)
}

// Dequeue removes and returns the highest-priority job from the heap
func (sq *SchedulerQueue) Dequeue() *model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	if sq.heap.Len() == 0 {
		return nil
	}
	job := heap.Pop(&sq.heap).(*model.Job)
	sq.tenantCounts[job.Tenant]--
	return job
}

What does this code do?

  • SchedulerQueue holds the heap, a jobs map for O(1) lookups by ID, per-tenant counters, tenant weights, and a pending slice for delayed jobs.
  • Every public method locks mu before touching shared state. This prevents data races when the HTTP handlers and scheduler goroutine run concurrently.
  • Enqueue adjusts the job's priority through effectivePriority (you will add this method in the next substep) before pushing onto the heap.

You won't see a change yet. The file won't compile until the fairness methods are added in the next substep.

  • Now add the query and promotion methods below the Dequeue function in the same file:
// GetJob returns a job by ID (nil if not found)
func (sq *SchedulerQueue) GetJob(id string) *model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()
	return sq.jobs[id]
}

// ListJobs returns jobs filtered by tenant and/or state
func (sq *SchedulerQueue) ListJobs(tenant, state string) []*model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	var result []*model.Job
	for _, job := range sq.jobs {
		if tenant != "" && job.Tenant != tenant {
			continue
		}
		if state != "" && string(job.State) != state {
			continue
		}
		result = append(result, job)
	}
	return result
}

// Stats returns queue metrics: depth, totals, per-tenant counts, state counts
func (sq *SchedulerQueue) Stats() map[string]any {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	stateCounts := make(map[string]int)
	for _, job := range sq.jobs {
		stateCounts[string(job.State)]++
	}

	return map[string]any{
		"queue_depth":   sq.heap.Len(),
		"total_jobs":    len(sq.jobs),
		"tenant_counts": sq.tenantCounts,
		"state_counts":  stateCounts,
	}
}

// PromotePending moves delayed jobs into the heap once their delay expires
func (sq *SchedulerQueue) PromotePending() []*model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	var promoted []*model.Job
	var remaining []*model.Job

	now := currentTime()
	for _, job := range sq.pending {
		if now.After(job.SubmittedAt.Add(job.Delay)) || now.Equal(job.SubmittedAt.Add(job.Delay)) {
			job.State = model.StateQueued
			job.Priority = sq.effectivePriority(job)
			sq.tenantCounts[job.Tenant]++
			heap.Push(&sq.heap, job)
			promoted = append(promoted, job)
		} else {
			remaining = append(remaining, job)
		}
	}
	sq.pending = remaining
	return promoted
}

What does this code do?

  • GetJob does a map lookup by job ID so the API can return a single job's details.
  • ListJobs iterates the full job map and filters by tenant and state. This supports the ?tenant= and ?state= query parameters you will add in the API step.
  • Stats returns a snapshot of queue health: how deep the heap is, how many jobs each tenant has, and how many jobs are in each state.
  • PromotePending scans delayed jobs and moves eligible ones into the heap. The scheduler goroutine will call this on every tick.

The file still won't compile. The next substep adds the two missing pieces.

Add weighted fair queueing logic

Without fairness, a tenant that submits 1000 jobs can starve a tenant with 1 job. Weighted fair queueing solves this by penalising tenants proportionally to how many in-flight jobs they already have.

How does the fairness formula work?

Each tenant has a weight (default 1). A job's effective priority is calculated as:

effectivePriority = basePriority + (tenantJobsInFlight / tenantWeight)

A tenant with 10 in-flight jobs and weight 1 gets +10 added to every new job's priority, pushing it down the queue. A tenant with weight 5 only gets +2, meaning higher-weight tenants are penalised less.

  • Add the fairness methods and the currentTime variable at the bottom of internal/queue/scheduler_queue.go:
// SetTenantWeight configures a tenant's fairness weight (higher = less penalty)
func (sq *SchedulerQueue) SetTenantWeight(tenant string, weight int) {
	sq.mu.Lock()
	defer sq.mu.Unlock()
	sq.tenantWeights[tenant] = weight
}

// effectivePriority adjusts a job's base priority by tenant load
func (sq *SchedulerQueue) effectivePriority(job *model.Job) int {
	weight := sq.tenantWeights[job.Tenant]
	if weight <= 0 {
		weight = 1
	}
	inFlight := sq.tenantCounts[job.Tenant]
	return job.Priority + (inFlight / weight)
}

// currentTime is a function variable for testability (swap with a fake clock in tests)
var currentTime = func() time.Now

What does this code do?

  • SetTenantWeight lets you configure how much fairness penalty a tenant receives. Higher weight means less penalty per in-flight job.
  • effectivePriority is unexported (lowercase) because only the queue itself should compute it. It defaults weight to 1 to avoid a divide-by-zero.
  • currentTime wraps time.Now in a function variable. This is a standard Go testing pattern that lets you inject a fake clock in tests without changing production code.
  • Save internal/queue/scheduler_queue.go.
  • Verify everything compiles by running:
go build ./...

You should see zero output, confirming no errors across both queue files.

Seeing an undefined error?

  • If you see sq.effectivePriority undefined, make sure you added the effectivePriority method in scheduler_queue.go (not a separate file). Methods must be in the same package as the struct they belong to.
  • If you see currentTime undefined, confirm the var currentTime line is at the bottom of scheduler_queue.go.
  • Double check that package queue is at the top of both priority_queue.go and scheduler_queue.go.

Help me debug my scheduler_queue.go compile error

✔️ Awesome, I've got everything!

Great. Confirm both files are saved in the internal/queue/ directory.

ⓧ I'd like to double check the full code

Here is the complete internal/queue/scheduler_queue.go:

package queue

import (
	"container/heap"
	"sync"

	"github.com/yourname/testscheduler/internal/model"
)

type SchedulerQueue struct {
	mu            sync.Mutex
	heap          JobHeap
	jobs          map[string]*model.Job
	tenantCounts  map[string]int
	tenantWeights map[string]int
	pending       []*model.Job
}

func NewSchedulerQueue() *SchedulerQueue {
	sq := &SchedulerQueue{
		jobs:          make(map[string]*model.Job),
		tenantCounts:  make(map[string]int),
		tenantWeights: make(map[string]int),
	}
	heap.Init(&sq.heap)
	return sq
}

func (sq *SchedulerQueue) Enqueue(job *model.Job) {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	job.State = model.StateQueued
	job.Priority = sq.effectivePriority(job)
	sq.jobs[job.ID] = job
	sq.tenantCounts[job.Tenant]++
	heap.Push(&sq.heap, job)
}

func (sq *SchedulerQueue) AddPending(job *model.Job) {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	job.State = model.StatePending
	sq.jobs[job.ID] = job
	sq.pending = append(sq.pending, job)
}

func (sq *SchedulerQueue) Dequeue() *model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	if sq.heap.Len() == 0 {
		return nil
	}
	job := heap.Pop(&sq.heap).(*model.Job)
	sq.tenantCounts[job.Tenant]--
	return job
}

func (sq *SchedulerQueue) GetJob(id string) *model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()
	return sq.jobs[id]
}

func (sq *SchedulerQueue) ListJobs(tenant, state string) []*model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	var result []*model.Job
	for _, job := range sq.jobs {
		if tenant != "" && job.Tenant != tenant {
			continue
		}
		if state != "" && string(job.State) != state {
			continue
		}
		result = append(result, job)
	}
	return result
}

func (sq *SchedulerQueue) Stats() map[string]any {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	stateCounts := make(map[string]int)
	for _, job := range sq.jobs {
		stateCounts[string(job.State)]++
	}

	return map[string]any{
		"queue_depth":   sq.heap.Len(),
		"total_jobs":    len(sq.jobs),
		"tenant_counts": sq.tenantCounts,
		"state_counts":  stateCounts,
	}
}

func (sq *SchedulerQueue) PromotePending() []*model.Job {
	sq.mu.Lock()
	defer sq.mu.Unlock()

	var promoted []*model.Job
	var remaining []*model.Job

	now := currentTime()
	for _, job := range sq.pending {
		if now.After(job.SubmittedAt.Add(job.Delay)) || now.Equal(job.SubmittedAt.Add(job.Delay)) {
			job.State = model.StateQueued
			job.Priority = sq.effectivePriority(job)
			sq.tenantCounts[job.Tenant]++
			heap.Push(&sq.heap, job)
			promoted = append(promoted, job)
		} else {
			remaining = append(remaining, job)
		}
	}
	sq.pending = remaining
	return promoted
}

func (sq *SchedulerQueue) SetTenantWeight(tenant string, weight int) {
	sq.mu.Lock()
	defer sq.mu.Unlock()
	sq.tenantWeights[tenant] = weight
}

func (sq *SchedulerQueue) effectivePriority(job *model.Job) int {
	weight := sq.tenantWeights[job.Tenant]
	if weight <= 0 {
		weight = 1
	}
	inFlight := sq.tenantCounts[job.Tenant]
	return job.Priority + (inFlight / weight)
}

var currentTime = func() time.Now

Your priority queue engine is complete. It orders jobs by urgency, breaks ties fairly, and protects shared state with a mutex. Next up, you will build the HTTP API that lets clients submit and query jobs through this queue.

Build the HTTP API

Your priority queue is built, tested, and compiling cleanly. Now you need a way for external systems to actually submit jobs into it and check their status.

In this step, you will build an HTTP API using Go 1.22+ method-based routing. The API layer gives your scheduler a network interface so clients can submit test jobs, query their state, and inspect queue statistics.

In this step, get ready to:
  • Set up a server struct with Go 1.22+ method-based routing and request logging middleware.
  • Implement a job submission handler with JSON validation and priority enforcement.
  • Implement query handlers for retrieving job details, listing jobs with filters, and viewing queue statistics.
Set up the HTTP server with Go 1.22+ routing

Go 1.22 introduced method-based routing patterns in the standard library's net/http package. This means you can register routes like "POST /jobs" directly on a ServeMux without any third-party router.

  • Create the file internal/api/server.go and add the server skeleton with routing, logging middleware, and ID generation:
package api

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"log"
	"net/http"
	"time"

	"github.com/yourname/testscheduler/internal/model"
	"github.com/yourname/testscheduler/internal/queue"
)

// Server holds a reference to the scheduler queue
type Server struct {
	queue *queue.SchedulerQueue
}

// NewServer creates a server wired to the given queue
func NewServer(q *queue.SchedulerQueue) *Server {
	return &Server{queue: q}
}

// Handler returns an http.Handler with all routes registered
func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("POST /jobs", s.withLogging(s.submitJob))
	mux.HandleFunc("GET /jobs/{id}", s.withLogging(s.getJob))
	mux.HandleFunc("GET /jobs", s.withLogging(s.listJobs))
	mux.HandleFunc("GET /stats", s.withLogging(s.getStats))
	return mux
}

// withLogging wraps a handler to log method, path, and duration
func (s *Server) withLogging(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		next(w, r)
		log.Printf("[API] %s %s %v", r.Method, r.URL.Path, time.Since(start))
	}
}

// generateID produces a random 32-character hex string
func generateID() string {
	b := make([]byte, 16)
	rand.Read(b)
	return hex.EncodeToString(b)
}

What does this code do?

The Server struct holds a pointer to your SchedulerQueue so every handler can enqueue, dequeue, and query jobs.

The Handler() method registers four routes using Go 1.22+ method-based patterns. Each route string includes the HTTP method as a prefix, so "POST /jobs" only matches POST requests to /jobs.

The withLogging middleware wraps each handler, recording how long the request took. The generateID helper uses crypto/rand to produce a cryptographically random 32-character hex string for job IDs.

Why Go 1.22+ routing instead of a third-party router?

Before Go 1.22, the standard ServeMux could only match path prefixes. Teams reached for packages like chi or gorilla/mux to get method-based routing and path parameters.

Go 1.22 added both features natively. Using the standard library keeps your project at zero external dependencies, which is exactly the constraint that teaches you the underlying patterns.

You won't see a change yet. The file references handler methods (submitJob, getJob, etc.) that you will write next.

Implement the job submission handler

The submission handler is the front door to your scheduler. It parses an incoming JSON body, validates the fields, generates a unique job ID, and routes the job into the queue.

  • Below the generateID function in internal/api/server.go, add the submitJob handler:
func (s *Server) submitJob(w http.ResponseWriter, r *http.Request) {
	// Define the expected request shape
	var req struct {
		Tenant    string `json:"tenant"`
		TestSuite string `json:"test_suite"`
		Priority  int    `json:"priority"`
		DelaySec  int    `json:"delay_sec,omitempty"`
	}

	// Decode the JSON body
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, `{"error":"invalid JSON"}`, http.StatusBadRequest)
		return
	}

	// Validate required fields
	if req.Tenant == "" || req.TestSuite == "" {
		http.Error(w, `{"error":"tenant and test_suite are required"}`, http.StatusBadRequest)
		return
	}
	if req.Priority < 0 || req.Priority > 10 {
		http.Error(w, `{"error":"priority must be 0-10"}`, http.StatusBadRequest)
		return
	}

	// Build the job with a unique ID and current timestamp
	job := &model.Job{
		ID:          generateID(),
		Tenant:      req.Tenant,
		TestSuite:   req.TestSuite,
		Priority:    req.Priority,
		State:       model.StatePending,
		Delay:       time.Duration(req.DelaySec) * time.Second,
		SubmittedAt: time.Now(),
	}

	// Route to pending list or directly to queue based on delay
	if req.DelaySec > 0 {
		s.queue.AddPending(job)
	} else {
		s.queue.Enqueue(job)
	}

	// Return 201 Created with the full job as JSON
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(job)
}

What does this handler do?

  • It decodes the request body into an anonymous struct. If the JSON is malformed, it returns a 400 immediately.
  • It validates that tenant and test_suite are non-empty and that priority is between 0 and 10.
  • If delay_sec is set, the job goes into the pending list (the scheduler will promote it later). Otherwise it enters the queue immediately.
  • The handler returns HTTP 201 with the full job object, so the client knows the assigned ID and current state.

You won't see a change yet. The file still references getJob, listJobs, and getStats in the route table. The next chunk adds those.

Implement the query handlers

Clients need to check on jobs after submitting them. These three handlers expose job details, filtered lists, and aggregate statistics.

  • Below the submitJob function in internal/api/server.go, add the three query handlers:
func (s *Server) getJob(w http.ResponseWriter, r *http.Request) {
	// Extract the job ID from the URL path parameter
	id := r.PathValue("id")
	job := s.queue.GetJob(id)
	if job == nil {
		http.Error(w, `{"error":"job not found"}`, http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(job)
}

func (s *Server) listJobs(w http.ResponseWriter, r *http.Request) {
	// Read optional query filters
	tenant := r.URL.Query().Get("tenant")
	state := r.URL.Query().Get("state")
	jobs := s.queue.ListJobs(tenant, state)
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(jobs)
}

func (s *Server) getStats(w http.ResponseWriter, r *http.Request) {
	// Return aggregate queue statistics
	stats := s.queue.Stats()
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(stats)
}

What do these handlers do?

  • getJob uses r.PathValue("id") to extract the job ID from the URL wildcard {id} registered in the route pattern. This is a Go 1.22 feature that replaces the need for third-party path parameter parsing.
  • listJobs reads optional ?tenant= and ?state= query parameters to filter the results. If neither is set, all jobs are returned.
  • getStats returns queue depth, total jobs, per-tenant counts, and jobs grouped by state. This gives operators a dashboard-ready overview.
  • Save internal/api/server.go.
  • Confirm everything compiles by running:
go build ./...

You should see no output and no errors. The build exits silently on success.

Seeing a compile error?

  • If you see undefined: s.submitJob or similar, make sure all three handler functions (submitJob, getJob, listJobs, getStats) are present in the same file with receiver (s *Server).
  • If you see an import error for model or queue, check that your import paths match your module name exactly: github.com/yourname/testscheduler/internal/model.
  • If you see unused import errors, make sure submitJob references model.StatePending and time.Now() so all imports are used.

Help me debug a compile error in my API server file.

✔️ Awesome, I've got everything!

Great. Double check you have saved internal/api/server.go and that go build ./... exits cleanly.

ⓧ I'd like to double check the full code

Here is the complete internal/api/server.go file. Compare your code against this:

package api

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"log"
	"net/http"
	"time"

	"github.com/yourname/testscheduler/internal/model"
	"github.com/yourname/testscheduler/internal/queue"
)

type Server struct {
	queue *queue.SchedulerQueue
}

func NewServer(q *queue.SchedulerQueue) *Server {
	return &Server{queue: q}
}

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("POST /jobs", s.withLogging(s.submitJob))
	mux.HandleFunc("GET /jobs/{id}", s.withLogging(s.getJob))
	mux.HandleFunc("GET /jobs", s.withLogging(s.listJobs))
	mux.HandleFunc("GET /stats", s.withLogging(s.getStats))
	return mux
}

func (s *Server) submitJob(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Tenant    string `json:"tenant"`
		TestSuite string `json:"test_suite"`
		Priority  int    `json:"priority"`
		DelaySec  int    `json:"delay_sec,omitempty"`
	}

	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, `{"error":"invalid JSON"}`, http.StatusBadRequest)
		return
	}

	if req.Tenant == "" || req.TestSuite == "" {
		http.Error(w, `{"error":"tenant and test_suite are required"}`, http.StatusBadRequest)
		return
	}
	if req.Priority < 0 || req.Priority > 10 {
		http.Error(w, `{"error":"priority must be 0-10"}`, http.StatusBadRequest)
		return
	}

	job := &model.Job{
		ID:          generateID(),
		Tenant:      req.Tenant,
		TestSuite:   req.TestSuite,
		Priority:    req.Priority,
		State:       model.StatePending,
		Delay:       time.Duration(req.DelaySec) * time.Second,
		SubmittedAt: time.Now(),
	}

	if req.DelaySec > 0 {
		s.queue.AddPending(job)
	} else {
		s.queue.Enqueue(job)
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(job)
}

func (s *Server) getJob(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	job := s.queue.GetJob(id)
	if job == nil {
		http.Error(w, `{"error":"job not found"}`, http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(job)
}

func (s *Server) listJobs(w http.ResponseWriter, r *http.Request) {
	tenant := r.URL.Query().Get("tenant")
	state := r.URL.Query().Get("state")
	jobs := s.queue.ListJobs(tenant, state)
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(jobs)
}

func (s *Server) getStats(w http.ResponseWriter, r *http.Request) {
	stats := s.queue.Stats()
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(stats)
}

func (s *Server) withLogging(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		next(w, r)
		log.Printf("[API] %s %s %v", r.Method, r.URL.Path, time.Since(start))
	}
}

func generateID() string {
	b := make([]byte, 16)
	rand.Read(b)
	return hex.EncodeToString(b)
}

Your API is compiled and ready to serve requests. Next up, you will wire it into a running HTTP server with a background scheduler loop and graceful shutdown.

Implement the Scheduler Loop and Wire Everything Together

Your queue is built and your API handles requests. The missing piece is the background process that actually drives the system forward.

In a real distributed system, a scheduler is the component that watches for work and pushes it through the pipeline on a regular cadence. You will build a ticker-based scheduler goroutine that promotes delayed jobs, then wire everything into a single main.go entrypoint with graceful shutdown so in-flight requests drain cleanly when you press Ctrl+C.

In this step, get ready to:
  • Create a ticker-based scheduler loop that promotes delayed jobs on a 500ms cadence.
  • Wire the queue, scheduler, and HTTP server together in a single entrypoint.
  • Implement OS signal handling for graceful shutdown with request draining.
Build the scheduler loop

The scheduler runs as a background goroutine on a fixed interval. Every 500 milliseconds it checks for pending jobs whose delay has elapsed and promotes them into the priority queue.

  • Create the file internal/scheduler/scheduler.go and paste in the following code:
package scheduler

import (
	"context"
	"log"
	"time"

	"github.com/yourname/testscheduler/internal/queue"
)

// Scheduler runs a ticker-based loop that promotes delayed jobs
type Scheduler struct {
	queue    *queue.SchedulerQueue
	interval time.Duration
}

// New creates a Scheduler with the given queue and tick interval
func New(q *queue.SchedulerQueue, interval time.Duration) *Scheduler {
	return &Scheduler{
		queue:    q,
		interval: interval,
	}
}

// Run starts the scheduler loop, exiting when ctx is canceled
func (s *Scheduler) Run(ctx context.Context) {
	ticker := time.NewTicker(s.interval)
	defer ticker.Stop()

	log.Printf("[SCHEDULER] started with interval=%v", s.interval)

	for {
		select {
		case <-ctx.Done():
			log.Println("[SCHEDULER] stopped")
			return
		case <-ticker.C:
			promoted := s.queue.PromotePending()
			for _, job := range promoted {
				log.Printf("[SCHEDULER] promoted job=%s tenant=%s delay=%v", job.ID, job.Tenant, job.Delay)
			}
		}
	}
}

What does this code do?

The Scheduler struct holds a reference to the queue and a configurable tick interval.

  • time.NewTicker creates a channel that fires every 500ms, giving the scheduler a steady heartbeat.
  • The select statement listens on two channels: the ticker (do work) and the context's Done channel (shut down).
  • Each tick calls PromotePending() on the queue, which moves delayed jobs whose wait time has elapsed from Pending into the heap as Queued.
  • When the context is canceled (e.g. via Ctrl+C), the loop exits cleanly and logs a stop message.

Why a ticker instead of time.AfterFunc?

A ticker provides a consistent heartbeat that is easy to reason about. You can compose it with context cancellation and extend it later with heartbeat checking or lease expiration.

Using time.AfterFunc per job would require tracking and canceling individual timers. The ticker approach scales linearly and keeps all promotion logic in one place.

  • Save internal/scheduler/scheduler.go.
  • Confirm the file compiles by running:
go build ./...

You should see no output, meaning zero compile errors.

Seeing an import error?

  • Make sure the file path is exactly internal/scheduler/scheduler.go. The package declaration must be package scheduler to match the directory name.
  • Confirm your import path uses github.com/yourname/testscheduler/internal/queue and matches the module name in your go.mod file.

Help me debug a Go import error in my scheduler package.

Wire everything together in main.go

The entrypoint needs to start the scheduler goroutine, start the HTTP server, and then wait for an OS signal to trigger a graceful shutdown sequence.

  • Open cmd/scheduler/main.go and replace the entire placeholder content with the following:
package main

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

	"github.com/yourname/testscheduler/internal/api"
	"github.com/yourname/testscheduler/internal/queue"
	"github.com/yourname/testscheduler/internal/scheduler"
)

func main() {
	// Create a context that cancels on SIGINT or SIGTERM
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	// Initialize shared queue, scheduler, and API server
	q := queue.NewSchedulerQueue()
	sched := scheduler.New(q, 500*time.Millisecond)
	server := api.NewServer(q)

	// Start the scheduler loop in a background goroutine
	go sched.Run(ctx)

	// Configure and start the HTTP server
	httpServer := &http.Server{
		Addr:    ":8080",
		Handler: server.Handler(),
	}

	go func() {
		log.Printf("[SERVER] listening on :8080")
		if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("[SERVER] failed to start: %v", err)
		}
	}()

	// Block until a shutdown signal arrives
	<-ctx.Done()
	log.Println("[SERVER] shutting down gracefully...")

	// Give in-flight requests 5 seconds to complete
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := httpServer.Shutdown(shutdownCtx); err != nil {
		log.Printf("[SERVER] forced shutdown: %v", err)
	}

	fmt.Println("[SERVER] shutdown complete")
}

What does this code do?

  • signal.NotifyContext creates a context that automatically cancels when the process receives SIGINT (Ctrl+C) or SIGTERM (a kill signal from the OS).
  • All three components (queue, scheduler, server) share the same SchedulerQueue instance, so the API writes jobs and the scheduler reads them.
  • go sched.Run(ctx) launches the ticker loop in its own goroutine. When ctx cancels, the loop exits.
  • The HTTP server runs in a separate goroutine. The main goroutine blocks on <-ctx.Done() until a signal arrives.
  • httpServer.Shutdown stops accepting new connections and waits up to 5 seconds for in-flight requests to finish before forcing a close.
  • Save cmd/scheduler/main.go.
  • Compile the entire project by running:
go build ./...

Zero output means no compile errors across all packages.

  • Start the server by running:
go run ./cmd/scheduler/

You should see two log lines confirming the scheduler and server are running:

[SCHEDULER] started with interval=500ms
[SERVER] listening on :8080
  • Open a second terminal and submit a test job by running:
curl -X POST http://localhost:8080/jobs -H "Content-Type: application/json" -d '{"tenant":"acme","test_suite":"unit-tests","priority":3}'

You should see a JSON response with a generated id and state set to queued.

  • Return to the terminal running the server and press Ctrl+C to trigger graceful shutdown.

You should see the shutdown sequence logged:

[SCHEDULER] stopped
[SERVER] shutting down gracefully...
[SERVER] shutdown complete

Server won't start or port already in use?

  • If you see an error about port 8080, another process is using it. On macOS/Linux, run lsof -i :8080 to find the process, then kill it.
  • If you see an import error for the scheduler package, confirm the directory is internal/scheduler/ and the package declaration is package scheduler.
  • If shutdown does not print the complete message, make sure you used Ctrl+C in the terminal where the server is running, not the curl terminal.

Help me debug my Go scheduler server startup.

Test a delayed job promotion

To prove the scheduler loop is working, you can submit a job with a delay and watch it get promoted from Pending to Queued after the delay elapses.

  • Start the server again by running:
go run ./cmd/scheduler/
  • In your second terminal, submit a job with a 3-second delay by running:
curl -X POST http://localhost:8080/jobs -H "Content-Type: application/json" -d '{"tenant":"acme","test_suite":"smoke-tests","priority":1,"delay_sec":3}'

The response shows state as pending. After roughly 3 seconds, check the server terminal. You should see:

[SCHEDULER] promoted job=<id> tenant=acme delay=3s
  • Verify the state changed by running:
curl http://localhost:8080/stats

You should see queue_depth of 1 and the state_counts showing queued instead of pending.

  • Press Ctrl+C to stop the server.

What just happened?

The job sat in Pending state for 3 seconds because of its delay. The scheduler's 500ms ticker checked it every half-second. Once the delay elapsed, PromotePending() moved it into the heap and flipped its state to Queued.

This is the same pattern production systems use for scheduled builds, retry backoff, and rate-limited task execution.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

Here is the full internal/scheduler/scheduler.go:

package scheduler

import (
	"context"
	"log"
	"time"

	"github.com/yourname/testscheduler/internal/queue"
)

type Scheduler struct {
	queue    *queue.SchedulerQueue
	interval time.Duration
}

func New(q *queue.SchedulerQueue, interval time.Duration) *Scheduler {
	return &Scheduler{
		queue:    q,
		interval: interval,
	}
}

func (s *Scheduler) Run(ctx context.Context) {
	ticker := time.NewTicker(s.interval)
	defer ticker.Stop()

	log.Printf("[SCHEDULER] started with interval=%v", s.interval)

	for {
		select {
		case <-ctx.Done():
			log.Println("[SCHEDULER] stopped")
			return
		case <-ticker.C:
			promoted := s.queue.PromotePending()
			for _, job := range promoted {
				log.Printf("[SCHEDULER] promoted job=%s tenant=%s delay=%v", job.ID, job.Tenant, job.Delay)
			}
		}
	}
}

Here is the full cmd/scheduler/main.go:

package main

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

	"github.com/yourname/testscheduler/internal/api"
	"github.com/yourname/testscheduler/internal/queue"
	"github.com/yourname/testscheduler/internal/scheduler"
)

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

	q := queue.NewSchedulerQueue()
	sched := scheduler.New(q, 500*time.Millisecond)
	server := api.NewServer(q)

	go sched.Run(ctx)

	httpServer := &http.Server{
		Addr:    ":8080",
		Handler: server.Handler(),
	}

	go func() {
		log.Printf("[SERVER] listening on :8080")
		if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("[SERVER] failed to start: %v", err)
		}
	}()

	<-ctx.Done()
	log.Println("[SERVER] shutting down gracefully...")

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := httpServer.Shutdown(shutdownCtx); err != nil {
		log.Printf("[SERVER] forced shutdown: %v", err)
	}

	fmt.Println("[SERVER] shutdown complete")
}

Your control plane is fully operational. The scheduler ticks in the background, the API accepts jobs, and everything shuts down cleanly on a signal. Next up, you will tackle the Secret Mission and add per-tenant rate limiting to protect the system from noisy neighbors.

Secret mission

Per-Tenant Token Bucket Rate Limiter

Your scheduler accepts jobs from any tenant without limits. A single tenant could flood the queue with thousands of jobs, starving everyone else before the weighted fair queueing even kicks in. Can you add admission control that stops abuse at the door?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally on your machine with no cloud services or paid APIs. There are no ongoing costs.

Resources you used:

  • The running HTTP server process on port :8080.
  • The testscheduler project folder on your Desktop.

Keep everything running

No action needed. Choose this if you're still actively building or plan to continue with Projects 2 and 3 in this series.

  • Your server continues to run and accept requests on :8080.
  • Your testscheduler folder stays on your Desktop.
  • Go remains installed on your machine.

Pause - I'll come back to this later

Stop the running server to free up the port, but keep your project files for later.

  • Switch to the terminal where the server is running.
  • Press Ctrl+C to send SIGINT. The graceful shutdown you built will drain in-flight requests and stop the scheduler.
  • You should see [SERVER] shutdown complete in the terminal output.
  • To resume later, navigate back to the testscheduler folder and run go run ./cmd/scheduler/.

Delete - I don't want to use this again

Remove all project files and stop the server completely.

  • Switch to the terminal where the server is running and press Ctrl+C to stop it.
  • Delete the project folder by running:

macOS

cd ~/Desktop
rm -rf testscheduler

Windows

cd ~\Desktop
rmdir /s /q testscheduler
  • Confirm the folder is gone by running ls (macOS) or dir (Windows) on your Desktop. You should no longer see testscheduler listed.

Nice Work!

Nice Work!

You just built a production-patterned control plane for a distributed test job scheduler. From a blank Go module to a running HTTP server with priority ordering, tenant fairness, and graceful shutdown, this is the kind of system that decides what runs, when, and for whom.

Here's what you learned:

  • Built a heap-based priority queue from scratch using Go's container/heap interface, with weighted fair queueing to prevent tenant starvation.
  • Created a REST API using Go 1.22+ method-based routing with zero external dependencies, handling job submission, status queries, and queue stats.
  • Implemented graceful shutdown with OS signal handling, context cancellation, and in-flight request draining.
  • Secret Mission: Added per-tenant rate limiting using the token bucket algorithm with concurrent-safe double-checked locking.

This control plane is the foundation for Projects 2 and 3, where you will add workers with lease-based coordination, heartbeats, fault tolerance, and autoscaling.

Ready to quiz yourself?