Go MQTT Vending Backend · Lesson 6 · 8 minutes

Bound message work; make overload visible

Move application work out of the MQTT receive callback without creating an unlimited backlog or hiding queue rejection.

🎯 One skill: reason about admission🧠 Predict overload behavior🧪 In-memory queue

❓ Start with the story

A burst of vending heartbeats arrives while workers are slow. The callback returns, but the telemetry queue is full. What happened to the packet, and what can the backend truthfully claim?

Pause before reading: separate MQTT delivery acknowledgement from the backend's queue decision.

📍 Files to implement in the codebase

internal/dispatch/pool.go is already present. Read it first: this lesson teaches its admission, worker, and shutdown behavior, then aligns the routing, pool coordination, and both process entrypoints with the listings below.

Path	Action	What you will learn or change
`internal/dispatch/pool.go`	Read and understand existing file	Fixed workers, bounded queue, immediate admission outcomes, counters, bounded drain
`internal/dispatch/receiver.go`	Review and align	Route by parsed topic kind and copy payload ownership before enqueue
`internal/dispatch/shutdown.go`	Review and align	Stop every pool before waiting under one shared deadline
`cmd/backend/main.go`	Replace with complete target file below	Wire Section 5A service to separate bounded pools and preserve MQTT during drain
`cmd/simulator/main.go`	Replace with complete target file below	Handle device commands through bounded work and drain before disconnect

The routing and shutdown files are already present in this checkout. Compare them with the complete listings below and preserve the same admission and error semantics.

📄 File overview: `pool.go` controls finite work

internal/dispatch/pool.go owns a fixed worker set and a bounded jobs channel. It limits simultaneous work and waiting work. It does not make queued messages durable, and it cannot ask MQTT to reverse an acknowledgement already sent.

MQTT callback
copy + submit

→

Finite queue
admit or reject

→

Fixed workers
run service

📥 `Submit` — short summary

Returns immediately with an explicit outcome: accepted, full, or stopping.

func (p *Pool) Submit(ctx context.Context, job Job) (Outcome, error) {
    if job == nil { return Stopping, errors.New("nil job") }
    if err := ctx.Err(); err != nil { return Stopping, err }

    p.mu.Lock()
    defer p.mu.Unlock()
    if !p.accepting { return Stopping, nil }
    select {
    case p.jobs <- job:
        p.accepted.Add(1)
        return Accepted, nil
    default:
        p.rejected.Add(1)
        return Full, nil
    }
}

This is the focused Submit method excerpt from Pool. The complete file defines the pool fields, Outcome values, worker loop, and shutdown behavior.

🧩 Details: why each branch exists

Reject nil or already-cancelled work before touching the queue.
The mutex makes the accepting check and channel send atomic with shutdown closing the channel.
The default branch means “no slot now”: this method never waits for capacity.
Accepted means queued in memory, not completed or saved to disk.
Full is an application overload decision. With automatic acknowledgement, returning it does not undo PUBACK or request redelivery.

⚖️ Choose a policy per traffic class

📈 Telemetry

If the business allows it, drop a stale heartbeat and count the rejection. Freshness will age out. Make the loss visible.

🧾 Sales and commands

Silently dropping a report or command is usually unacceptable. Return an explicit failure and add durable admission before claiming crash-safe processing.

Separate pools can prevent telemetry from consuming every waiting slot reserved for important work. That is capacity isolation, not durability. Shutdown order matters too: stop producers, stop admission, drain workers to a deadline while needed transport remains alive, then disconnect.

🛑 What does `ctx.Err()` contain during shutdown?

The ctx passed to Pool.Shutdown is the drain context. If its deadline expires before workers finish, ctx.Err() returns context.DeadlineExceeded. If something cancels it earlier, it returns context.Canceled. If workers finish first, Pool.Shutdown returns nil instead. On timeout, the pool uses that context error as its shutdown result and cancels the worker job context. Cancellation is cooperative, so a handler must check its context or use context-aware I/O to stop promptly.

🧭 File and function map

File overview: Section 6 adds three pieces: pool.go owns bounded execution, receiver.go routes each message to the right pool, and shutdown.go closes all pools before waiting for their workers. The backend entrypoint below wires those pieces to the Section 5A service and existing MQTT inbox.

📄 `internal/dispatch/pool.go`

New summary: starts a fixed worker count and a queue with a fixed capacity.

Submit summary: reports Accepted, Full, or Stopping without waiting for queue room.

StopAdmission / Shutdown summary: close the submission path, wait for workers up to a deadline, then cancel remaining work.

Stats / worker summary: expose admission and completion counts; workers run jobs and count success or failure.

🔎 Details

The mutex protects both the accepting flag and channel close, preventing a send-on-closed-channel race. The nonblocking select makes overload immediate. Closing the jobs channel lets workers finish buffered jobs; the shared deadline keeps draining bounded. If it expires, cancellation is cooperative: a job that ignores its context can keep running. Counters describe process-local events, not durable business records.

📄 `internal/dispatch/receiver.go`

Receiver.Submit summary: parses the topic, selects the telemetry or important pool, copies payload bytes, and submits a closure that calls the application processor.

🔎 Details

Routing on a parsed Kind keeps traffic policy out of the MQTT callback. Commands, sales, and command responses use the important pool; only telemetry uses the telemetry pool. Copying the payload transfers ownership: the worker no longer depends on Paho or a reused inbox buffer. The job closure captures that owned copy and runs later with the worker's context.

📄 `internal/dispatch/shutdown.go`

Shutdown summary: stops submissions to every pool first, then drains them using one shared deadline.

🔎 Details

Stopping all pools before waiting prevents one pool from accepting new work while the other drains. Reusing one context bounds the total shutdown time instead of allowing a fresh timeout for each pool. errors.Join preserves timeout failures from multiple pools. Keep MQTT alive during this interval if jobs need to publish their result; disconnect only after the drain attempt.

🔌 Integration overview: edit `cmd/backend/main.go`

run summary: builds the service and pools once, starts MQTT, sends each inbox message through Receiver, and drains workers when the signal context ends.

Details: the signal context stops the main loop, while a separate transport context keeps MQTT available during the bounded drain. The signal case itself only logs and returns; as run returns, deferred cleanup executes in reverse registration order. The dispatcher drain defer was registered after mqtt.Stop, so it runs first. Pool jobs use an independent parent context so the signal does not cancel them before draining. Once drain completes or times out, mqtt.Stop disconnects the client. The inbox and callback remain bounded; if more messages arrive during shutdown they can still be dropped under automatic acknowledgement, which this lesson reports rather than hiding.

🧩 Complete Section 6 backend wiring: cmd/backend/main.go

This is the full target entrypoint layered on the Section 5A service. The three dispatcher files follow it below.

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"syscall"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/deviceview"
	"example.com/mqtt5-learning/internal/dispatch"
	"example.com/mqtt5-learning/internal/mqtt"
)

func main() {
	if err := run(); err != nil {
		log.Fatal(err)
	}
}

func run() error {
	cfg, err := config.Load(os.Args[1:])
	if err != nil {
		return fmt.Errorf("failed to load config: %w", err)
	}

	lifecycleCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	transportCtx, stopTransport := context.WithCancel(context.Background())
	defer stopTransport()

	store := deviceview.NewStore()
	service, err := deviceview.NewService(store, slog.Default(), 90*time.Second)
	if err != nil {
		return fmt.Errorf("create device-view service: %w", err)
	}

	telemetryFilter, err := contracts.DeviceFilter(contracts.KindTelemetry)
	if err != nil { return err }
	responsesFilter, err := contracts.DeviceFilter(contracts.KindResponses)
	if err != nil { return err }
	commandTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindCommands)
	if err != nil { return err }
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(transportCtx, cfg, []contracts.TopicFilter{telemetryFilter, responsesFilter}, inbox)
	if err != nil {
		return fmt.Errorf("failed to start MQTT client: %w", err)
	}
	defer mqtt.Stop(manager)

	telemetryPool := dispatch.New(context.Background(), 2, 16)
	importantPool := dispatch.New(context.Background(), 2, 32)
	receiver := &dispatch.Receiver{
		Telemetry: telemetryPool,
		Important: importantPool,
		Process:   service.HandleMessage,
	}
	defer func() {
		drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		started := time.Now()
		if err := dispatch.Shutdown(drainCtx, telemetryPool, importantPool); err != nil {
			log.Printf("dispatcher shutdown incomplete: %v", err)
		}
		log.Printf("dispatcher drain finished elapsed=%s telemetry=%+v important=%+v",
			time.Since(started), telemetryPool.Stats(), importantPool.Stats())
	}()

	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	runID := time.Now().UnixNano()
	sequence := 0

	for {
		select {
		case <-lifecycleCtx.Done():
			log.Print("shutting down; draining accepted work before MQTT disconnect")
			return nil
		case message := <-inbox:
			outcome, submitErr := receiver.Submit(lifecycleCtx, message)
			if submitErr != nil || outcome != dispatch.Accepted {
				slog.Warn("message admission rejected", "topic", message.Topic, "outcome", outcome, "error", submitErr)
			}
		case <-ticker.C:
			sequence++
			id := fmt.Sprintf("ping-%d-%d", runID, sequence)
			payload, marshalErr := json.Marshal(contracts.Command{
				SchemaVersion: contracts.SchemaVersion,
				MessageID: id,
				DeviceID: cfg.MQTT.DeviceID,
				OccurredAt: time.Now().UTC().Format(time.RFC3339),
				CommandID: id,
				Command: "ping",
			})
			if marshalErr != nil {
				return fmt.Errorf("failed to marshal payload: %w", marshalErr)
			}
			if publishErr := mqtt.Publish(lifecycleCtx, manager, commandTopic, payload); publishErr != nil {
				log.Printf("command publication failed or outcome uncertain: %v", publishErr)
			}
		}
	}
}

🤖 Simulator integration overview: edit `cmd/simulator/main.go`

run summary: starts the device client, publishes heartbeats, and submits received commands to bounded workers instead of handling them inline in the main loop.

respond summary: rejects retained or misrouted commands, validates the contract, and publishes the pong using the worker context.

Details: the simulator attempts to drain accepted command jobs before closing MQTT so a worker can still publish its response. The drain has a five-second limit; if it expires, remaining in-memory work may not complete. A full important queue is logged as an application rejection; automatic acknowledgement may already have completed the subscriber delivery, so this RAM-only example cannot promise redelivery.

🧩 Complete Section 6 simulator wiring: cmd/simulator/main.go

This complete target file carries the Section 5 contract-aware simulator forward and moves inbound command work to the dispatcher.

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync/atomic"
	"syscall"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/dispatch"
	"example.com/mqtt5-learning/internal/mqtt"
	"github.com/eclipse/paho.golang/autopaho"
)

func main() {
	if err := run(); err != nil {
		log.Fatal(err)
	}
}

func run() error {
	cfg, err := config.Load(os.Args[1:])
	if err != nil {
		return fmt.Errorf("load config: %w", err)
	}
	lifecycleCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	transportCtx, stopTransport := context.WithCancel(context.Background())
	defer stopTransport()

	commandFilter, err := contracts.DeviceSubscription(cfg.MQTT.DeviceID, contracts.KindCommands)
	if err != nil { return err }
	telemetryTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindTelemetry)
	if err != nil { return err }
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(transportCtx, cfg, []contracts.TopicFilter{commandFilter}, inbox)
	if err != nil {
		return fmt.Errorf("start MQTT: %w", err)
	}
	defer mqtt.Stop(manager)

	runID := time.Now().UnixNano()
	telemetryPool := dispatch.New(context.Background(), 2, 16)
	importantPool := dispatch.New(context.Background(), 2, 16)
	var resultSequence atomic.Uint64
	receiver := &dispatch.Receiver{
		Telemetry: telemetryPool,
		Important: importantPool,
		Process: func(jobCtx context.Context, message mqtt.Message) error {
			resultID := fmt.Sprintf("result-%d-%d", runID, resultSequence.Add(1))
			return respond(jobCtx, manager, cfg.MQTT.DeviceID, resultID, message)
		},
	}
	defer func() {
		drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		started := time.Now()
		if err := dispatch.Shutdown(drainCtx, telemetryPool, importantPool); err != nil {
			log.Printf("dispatcher shutdown incomplete: %v", err)
		}
		log.Printf("dispatcher drain finished elapsed=%s telemetry=%+v important=%+v",
			time.Since(started), telemetryPool.Stats(), importantPool.Stats())
	}()

	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()
	sequence := uint64(0)
	for {
		select {
		case <-lifecycleCtx.Done():
			log.Print("shutting down; draining accepted command work before MQTT disconnect")
			return nil
		case <-ticker.C:
			sequence++
			heartbeat := contracts.Heartbeat{
				SchemaVersion: contracts.SchemaVersion,
				MessageID:     fmt.Sprintf("heartbeat-%d-%d", time.Now().Unix(), sequence),
				DeviceID:      cfg.MQTT.DeviceID,
				OccurredAt:    time.Now().UTC().Format(time.RFC3339),
			}
			payload, marshalErr := json.Marshal(heartbeat)
			if marshalErr != nil {
				return fmt.Errorf("encode heartbeat: %w", marshalErr)
			}
			if publishErr := mqtt.Publish(lifecycleCtx, manager, telemetryTopic, payload); publishErr != nil {
				log.Printf("heartbeat publish: %v", publishErr)
			}
		case message := <-inbox:
			outcome, submitErr := receiver.Submit(lifecycleCtx, message)
			if submitErr != nil || outcome != dispatch.Accepted {
				log.Printf("command admission rejected outcome=%d topic=%q error=%v", outcome, message.Topic, submitErr)
			}
		}
	}
}

func respond(ctx context.Context, manager *autopaho.ConnectionManager, deviceID, resultMessageID string, message mqtt.Message) error {
	if message.Retain {
		return fmt.Errorf("retained command rejected")
	}
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return err
	}
	if topic.Kind != contracts.KindCommands || topic.DeviceID != deviceID {
		return fmt.Errorf("command topic is not authorized for this device")
	}
	command, err := contracts.Decode[contracts.Command](message.Payload, topic, time.Now().UTC())
	if err != nil {
		return err
	}
	result := contracts.CommandResult{
		SchemaVersion: contracts.SchemaVersion,
		MessageID:     resultMessageID,
		DeviceID:      deviceID,
		OccurredAt:    time.Now().UTC().Format(time.RFC3339),
		CommandID:     command.CommandID,
		Status:        "completed",
	}
	payload, err := json.Marshal(result)
	if err != nil {
		return fmt.Errorf("encode result: %w", err)
	}
	responseTopic, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	if err != nil { return fmt.Errorf("build response topic: %w", err) }
	return mqtt.Publish(ctx, manager, responseTopic, payload)
}

🧪 Check your model

🛠️ Your tiny practice

When capacity is full, write one sentence and one metric for each class: heartbeat, sales report, command result. Then state the exact loss window created by automatic acknowledgement plus an in-memory queue. Compare with Section 6's dispatcher and shutdown code ↗.

📦 Section 6 complete dispatcher files

Complete source files for the added dispatcher package. The complete backend integration appears immediately above.

📄 internal/dispatch/pool.go

package dispatch

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
)

type Outcome uint8

// Outcomes make queue admission success and overload visible to callers.
const (
	Accepted Outcome = iota
	Full
	Stopping
)

type Job func(context.Context) error

type Stats struct {
	Accepted  uint64
	Rejected  uint64
	Completed uint64
	Failed    uint64
}

// Pool owns a bounded queue, worker lifecycle, and admission counters.
type Pool struct {
	mu        sync.Mutex
	accepting bool
	jobs      chan Job
	ctx       context.Context
	cancel    context.CancelFunc
	workers   sync.WaitGroup
	done      chan struct{}

	accepted  atomic.Uint64
	rejected  atomic.Uint64
	completed atomic.Uint64
	failed    atomic.Uint64
}

// New starts a fixed worker set and ties job cancellation to parent.
func New(parent context.Context, workers, capacity int) *Pool {
	if workers < 1 {
		workers = 1
	}
	if capacity < 0 {
		capacity = 0
	}
	ctx, cancel := context.WithCancel(parent)
	p := &Pool{accepting: true, jobs: make(chan Job, capacity), ctx: ctx, cancel: cancel, done: make(chan struct{})}
	p.workers.Add(workers)
	for i := 0; i < workers; i++ {
		go p.worker()
	}
	go func() { p.workers.Wait(); close(p.done) }()
	return p
}

// Submit never waits for queue capacity. The caller gets an explicit Full result.
// The caller context can cancel admission before it enters the queue.
func (p *Pool) Submit(ctx context.Context, job Job) (Outcome, error) {
	if job == nil {
		return Stopping, errors.New("nil job")
	}
	if err := ctx.Err(); err != nil {
		return Stopping, err
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if !p.accepting {
		return Stopping, nil
	}
	select {
	case p.jobs <- job:
		p.accepted.Add(1)
		return Accepted, nil
	default:
		p.rejected.Add(1)
		return Full, nil
	}
}

func (p *Pool) StopAdmission() {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.accepting {
		p.accepting = false
		close(p.jobs)
	}
}

// Shutdown stops new submissions and drains queued work until ctx expires.
// Jobs must observe their context so deadline cancellation can stop slow work.
func (p *Pool) Shutdown(ctx context.Context) error {
	p.StopAdmission()
	select {
	case <-p.done:
		p.cancel()
		return nil
	case <-ctx.Done():
		p.cancel()
		return ctx.Err()
	}
}

func (p *Pool) Stats() Stats {
	return Stats{Accepted: p.accepted.Load(), Rejected: p.rejected.Load(), Completed: p.completed.Load(), Failed: p.failed.Load()}
}

// Workers execute admitted jobs and count their final result.
func (p *Pool) worker() {
	defer p.workers.Done()
	for {
		select {
		case <-p.ctx.Done():
			return
		case job, ok := <-p.jobs:
			if !ok {
				return
			}
			if err := job(p.ctx); err != nil {
				p.failed.Add(1)
				continue
			}
			p.completed.Add(1)
		}
	}
}

📄 internal/dispatch/receiver.go

package dispatch

import (
	"context"
	"fmt"

	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/mqtt"
)

type Processor func(context.Context, mqtt.Message) error

// Receiver validates the route, selects a pool, and copies payload ownership.
type Receiver struct {
	Telemetry *Pool
	Important *Pool
	Process   Processor
}

func (r *Receiver) Submit(ctx context.Context, message mqtt.Message) (Outcome, error) {
	if r.Process == nil {
		return Stopping, fmt.Errorf("message processor is required")
	}
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return Stopping, err
	}
	pool := r.Telemetry
	if topic.Kind == contracts.KindCommands || topic.Kind == contracts.KindResponses || topic.Kind == contracts.KindSales {
		pool = r.Important
	}
	if pool == nil {
		return Stopping, fmt.Errorf("no queue configured for topic kind %q", topic.Kind)
	}
	copyOfMessage := message
	copyOfMessage.Payload = append([]byte(nil), message.Payload...)
	return pool.Submit(ctx, func(jobCtx context.Context) error { return r.Process(jobCtx, copyOfMessage) })
}

📄 internal/dispatch/shutdown.go

package dispatch

import (
	"context"
	"errors"
)

func Shutdown(ctx context.Context, pools ...*Pool) error {
	// Close admission on every pool before waiting for any worker to drain.
	for _, pool := range pools {
		if pool != nil {
			pool.StopAdmission()
		}
	}
	var result error
	// Share one deadline across the drain so total shutdown time stays bounded.
	for _, pool := range pools {
		if pool != nil {
			result = errors.Join(result, pool.Shutdown(ctx))
		}
	}
	return result
}

🏗️ Build and smoke-run the updated processes

After saving all three internal/dispatch files and wiring both entrypoints, run these commands from the repository root:

go test ./...
go test -race ./...
go build -o /tmp/mqtt5-lab-backend ./cmd/backend
go build -o /tmp/mqtt5-lab-simulator ./cmd/simulator

The race detector can report races only on paths the tests execute; it does not prove every worker schedule safe. The builds confirm both programs use the updated wiring.

Start only the named lab broker, then run the two binaries in separate terminals:

docker compose -p mqtt5-learning up -d broker
docker compose -p mqtt5-learning ps

/tmp/mqtt5-lab-simulator -username simulator -client-id lab-simulator-device-001

/tmp/mqtt5-lab-backend -username backend -client-id lab-backend

Confirm heartbeats and ping responses still complete while passing through worker queues. On Ctrl+C, observe the shutdown summary and compare accepted, rejected, completed, failed, and drain-time values if your wiring logs them. This ordinary traffic is a smoke run, not an overload test; use a finite burst larger than active workers plus queue capacity to observe Full outcomes. Stop the clients, then stop the lab broker with docker compose -p mqtt5-learning stop broker.

Ask me a follow-up if the counters or shutdown behavior differ from your prediction.

✅ Takeaway

A bounded queue protects process memory; an explicit overload policy protects business meaning. Neither makes work durable.

Tell me which message class you would admit first and why.

Bound message work; make overload visible

📍 Files to implement in the codebase

📄 File overview: `pool.go` controls finite work

📥 `Submit` — short summary

🧩 Details: why each branch exists

⚖️ Choose a policy per traffic class

📈 Telemetry

🧾 Sales and commands

🛑 What does ctx.Err() contain during shutdown?

🧭 File and function map

📄 internal/dispatch/pool.go

🔎 Details

📄 internal/dispatch/receiver.go

🔎 Details

📄 internal/dispatch/shutdown.go

🔎 Details

🔌 Integration overview: edit cmd/backend/main.go

🤖 Simulator integration overview: edit cmd/simulator/main.go

🧪 Check your model

🛠️ Your tiny practice

📄 internal/dispatch/pool.go

📄 internal/dispatch/receiver.go

📄 internal/dispatch/shutdown.go

🏗️ Build and smoke-run the updated processes

✅ Takeaway

🛑 What does `ctx.Err()` contain during shutdown?

📄 `internal/dispatch/pool.go`

📄 `internal/dispatch/receiver.go`

📄 `internal/dispatch/shutdown.go`

🔌 Integration overview: edit `cmd/backend/main.go`

🤖 Simulator integration overview: edit `cmd/simulator/main.go`