mqtt-5-learning/teaching/course-source/06-concurrency-cancellation.md

# 🧵 Section 6 — Manage callbacks, concurrency, and cancellation

## 🎬 Start with the system story

> **❓ Question:** **A burst of broker-forwarded messages arrives while the backend is doing slow work. Who is receiving packets, who is doing business work, and what happens when the application cannot keep up?**

Paho's network/client machinery receives packets and invokes configured handlers. In this lesson, a **callback** means the application function that Paho calls when a matching incoming PUBLISH reaches the client. The callback runs as part of Paho's receive path; it is not the broker, and it is not a background worker. Your application decides how work is scheduled, bounded, cancelled, and acknowledged. If a callback blocks or a queue fills, that can affect unrelated deliveries; if a message is acknowledged before durable handling and then dropped from RAM, MQTT cannot recover that application copy by itself.

```mermaid
flowchart LR
    B[Broker] -->|PUBLISH| P[Paho receive path]
    P --> C[Short callback copies/enqueues]
    C --> Q[Bounded application queue]
    Q --> W[Worker handles and persists]
    Q -->|full| O[Explicit overload policy]
    W --> A[Application outcome]
```

**Before each implementation step, ask:** What work must happen before the MQTT acknowledgement, what can happen afterward, and what is the defined behavior when capacity is exhausted? Follow [Lesson 0](00-system-lifecycle.md) for the broker-to-callback-to-application path.

Prerequisite: Section 5 validates messages, and Section 5A applies them through an application service to an in-memory device view. This lesson moves that same service behind workers. The backend and simulator continue to read their shared disposable password from `configs/lab.toml`; select each broker username with `-username` when starting the process.

Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. Save the complete source blocks at the project-relative paths shown below. This lesson includes the full dispatcher implementation; the exercises ask you to run it and inspect its admission, overload, and shutdown behavior.

## ✅ Understand the acknowledgement boundary before adding workers

### 📥 The publisher sends a message to the broker

For QoS 1, the publisher sends `PUBLISH` to the broker. The broker sends `PUBACK` to the publisher to complete that delivery hop. This says nothing about whether subscribers received or processed the message.

### 📬 The broker delivers to the backend

When the backend is a subscriber, the broker sends it a separate `PUBLISH`. Paho receives the packet and calls the registered callback. In automatic acknowledgement mode, Paho sends `PUBACK` to the broker after the callback returns.

If the callback enqueues the message and returns, the backend can acknowledge it before a worker processes it. That worker may later reject the message or the process may crash before handling it.

### ⚠️ What if the callback returns an error?

| Paho mode | What Paho does after the callback returns an error | Does the error request redelivery? |
| --- | --- | --- |
| Automatic acknowledgement | Sends `PUBACK` for QoS 1 after the callback returns | No. The error is local; it does not ask the broker to resend. |
| Manual acknowledgement | Does not automatically send `PUBACK` | No. Your code must call `Ack` when its policy says to acknowledge. |

In manual mode, leaving the message unacknowledged keeps that delivery in flight and can slow later QoS 1/2 deliveries. It does not itself guarantee a retry on the same connection. The broker may redeliver after a disconnect only if session, QoS, expiry, and broker state allow recovery.

For QoS 0, there is no `PUBACK`. QoS 2 has a multi-packet handshake, but completing that handshake still does not prove the subscriber's business action succeeded.

**❓ Question:** **The broker delivers a sales report with quantity `-3`. Can MQTT acknowledge it even if a worker later rejects it?**

**💬 Answer:** Yes. With automatic acknowledgement, Paho can send `PUBACK` after the callback returns, before the worker checks the report. The worker may then reject the sale. Record that application result separately; `PUBACK` does not mean the sale was accepted.

Trace the early QoS 1 implementation:

```text
publisher --QoS 1--> broker --QoS 1--> this subscriber's client
                              └-------> other subscriber clients independently
this client receives PUBLISH → callback copies into RAM queue → callback returns → sends PUBACK
                                                                       └→ worker validates/applies effect later
```

The worker may finish before or after the ACK. If the process crashes after the ACK but before the effect, the broker can consider its delivery complete while the application loses the work. Increasing channel capacity increases the possible backlog; it does not close that gap.

Read the arrows as two independent delivery hops. The publisher's QoS 1 exchange ends when the broker acknowledges that publisher. The broker then starts a separate delivery to this subscriber, and other subscribers receive their own independent deliveries. The callback's copy puts bytes into process memory; returning from that callback is the event that permits automatic acknowledgement in this example. The worker's later validation and side effect belong to the application lifecycle, not to the completed MQTT exchange. This is why queue admission and business completion need separate evidence and, for important work, separate durable records.

In manual mode, `EnableManualAcknowledgment` means Paho does not automatically acknowledge a QoS 1 message when the callback returns. Your application must call `received.Client.Ack(received.Packet)` to release it. Returning an error alone does not acknowledge or reject it, and does not request a retry. Unless your code calls `Ack`, that delivery stays unacknowledged on the connection. This can occupy an in-flight slot and slow later QoS 1/2 deliveries. For important work, the intended sequence is receive → validate → commit durable admission → call `Ack`. Preserve the receiving client and original packet for this operation; a payload-only copy is insufficient. Check Ack's error and study the pinned source before implementing it. Do not reuse this QoS 1 explanation as a claim that every QoS 2 handshake stage waits for your database.

The client's tracker sends acknowledgements in receipt order. If work 3 finishes before 1, acknowledging 3 can remain buffered behind 1. Slow work therefore consumes inflight capacity even with several workers. Manual acknowledgement delays protocol completion; it does not persist your Go queue, remove duplicate effects, or automatically schedule retries while the same connection stays open.

The practical choice is about the promise made to the broker. Automatic acknowledgement is simple, but returning after an in-memory enqueue accepts a possible loss window. Manual acknowledgement lets the application delay completion until its chosen checkpoint, but the checkpoint should be durable if a process crash must not lose work. Because acknowledgements are tracked in receipt order, finishing jobs concurrently does not guarantee that later packets can be acknowledged ahead of an earlier one. Keep that protocol limit in mind when choosing worker parallelism and Receive Maximum.

### ↩️ Choose what the callback promises

| Message class | Reasonable admission choice | Meaning of a successful admission |
| --- | --- | --- |
| Disposable heartbeat | Bounded RAM queue, counted drops | A recent sample may be processed while this process survives |
| Important report | Durable inbox before acknowledgement, introduced in Section 13 | Work can be recovered after this process restarts |
| Command | Validate identity/deadline and record dedup/admission state | Device acceptance still needs an explicit application result |

For this in-memory lesson, keep the loss limitation visible. Use a fake acknowledgement recorder to test ordering of admission and acknowledgement without pretending that the fake verifies broker behavior.

Use the table as a policy map, not as a claim that all message types need the same storage. A heartbeat is usually useful as a recent sample, so bounded memory plus a drop counter can be an acceptable contract. A report that drives accounting or inventory needs durable admission before the broker delivery is acknowledged; Section 13 builds that inbox. A command needs identity, expiry/deadline, and duplicate handling before the backend can claim it was accepted. Even then, admission is not device acceptance: the device must return an application-level result. A fake recorder can verify that code calls its acknowledgement hook only after the selected admission step; broker/session behavior still needs an integration experiment.

### 🧪 Predict the overload

Two workers taking 100 ms per item can complete about 20 items/second before other overhead. A burst of 100 items arriving much faster cannot fit into two busy workers plus 16 waiting slots. Predict rejection, then measure actual accepted/completed/rejected counts; scheduler timing affects the exact split. Increasing workers may reorder effects for the same device. If revision order matters, serialize that device's work or enforce revision checks.

The estimate follows from service time: each worker can process about ten 100 ms jobs per second, so two workers provide roughly twenty per second when work is uniformly sized. With both workers busy and a queue capacity of 16, at most 18 jobs can be active or waiting at once; additional arrivals encounter `Full` until a worker frees capacity. A fast finite burst may therefore produce a different accepted count depending on scheduling and exactly when arrivals occur. The counters tell you what happened in that run. More workers increase throughput but also allow operations for one device to overlap; use per-device serialization or revision checks when order is part of the data contract.

Shutdown also has a dependency: if draining work must publish results, keep its MQTT connection alive while draining. Stop new admission first, drain accepted work with a deadline, then disconnect transport and join workers. If you disconnect first, record unsent results for recovery; an in-memory-only lesson cannot promise to deliver them. The Section 4 baseline uses one signal context for AutoPaho; the complete Section 6 integration below separates signal and transport lifetimes so it can drain first.

Treat shutdown as an ordering protocol among producers, queues, workers, and transport. First stop the receive loop from submitting new jobs; then close pool admission. Give workers a bounded interval to finish while MQTT remains available for replies. Only after the drain should that transport be disconnected. If the deadline expires, cancellation asks cooperative handlers to stop, but unfinished in-memory jobs may be lost. Keep signal, drain, and transport lifetimes separate.

<details>
<summary><strong>❓ Question: What does a callback error do to QoS 1 acknowledgement?</strong></summary>

**💬 Answer:**

- **Automatic mode:** Paho sends `PUBACK` after the callback returns, even when it returns an error. The error does not request redelivery.
- **Manual mode:** Paho does not send `PUBACK` just because the callback returns. Your application must call `Ack`; otherwise the delivery remains in flight.

In neither mode does returning an error tell the broker to reject or retry the message.

</details>

## ↩️ 1. Trace callback and acknowledgement order

> **❓ Question:** **Does returning from the receive callback mean the business operation is durable, or only that the MQTT client may acknowledge delivery?**

Read the pinned source and write the observed sequence down before choosing a queue or acknowledgement policy.

Read `references/paho.golang-v0.23.0/paho/client.go`, especially receive callback processing, `EnableManualAcknowledgment`, and `Ack`. Write the observed order in `journal/06-concurrency.md`. Inspect this pinned source before choosing an acknowledgement policy.

Section 4 copies payloads into a bounded channel and drops messages when full. With automatic acknowledgement, Paho can send PUBACK for this client's delivery after the callback returns—even if the application later discovers that the queue was full. This is a deliberate loss window, not reliable ingestion.

For this source-reading step, follow the packet from the receive handler to callback invocation and then to the acknowledgement decision. Record which condition enables manual mode, whether callback errors change the automatic-ack path, and what `Ack` needs to identify the original delivery. These observations explain why a channel send is an application decision rather than an MQTT acknowledgement. The written notes become a reference for the implementation choices in the next section; do not infer behavior from a method name alone when the pinned client source is available.

## 🛠️ 2. Build a bounded dispatcher

> **❓ Question:** **When messages arrive faster than handlers can finish, where is excess work stored and how does the application report that capacity is exhausted?**

Set a finite capacity and make admission outcomes explicit for each message class.

### 🔍 `pool.go` — file overview

This file implements one bounded job queue and a fixed set of workers. Callers submit jobs without waiting for free capacity, inspect admission and processing counters, then stop submissions and drain work during shutdown. The pool limits waiting jobs and simultaneous jobs; it does not save them to disk.

Save this complete file as `internal/dispatch/pool.go`.

```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)
		}
	}
}
```

#### 🧩 Types at a glance

- `Outcome` reports whether a job was `Accepted`, rejected as `Full`, or refused because the pool is `Stopping`.
- `Job` is the function a worker runs. Its context lets the job observe pool cancellation.
- `Stats` reports admission counts separately from worker results: accepted, rejected, completed, and failed.
- `Pool` owns the queue, workers, lifecycle context, synchronization, and counters.

#### 🚀 `New` — start the pool

**Summary:** Creates the bounded queue and starts the requested number of workers.

**Details:** Worker counts below one become one; negative capacity becomes zero. The derived context connects parent cancellation to jobs. `done` closes when every worker exits, allowing shutdown to wait for the whole worker set. A zero-capacity channel admits a job only when a worker can receive it immediately; a positive capacity holds that many waiting jobs.

#### 📥 `Submit` — attempt immediate admission

**Summary:** Validates the request and either enqueues the job immediately or returns a reason it was not admitted.

**Details:** A nil job or already-cancelled caller context returns an error. Under the mutex, the method rejects submissions after shutdown begins. The `select` with a `default` case never waits for queue space: a successful send increments `Accepted`; no available slot increments `Rejected` and returns `Full`. The caller context is checked only before admission; the job itself receives the pool context when a worker runs it.

#### 🛑 `StopAdmission` — close submissions safely

**Summary:** Prevents future submissions and closes the job channel once.

**Details:** The mutex is shared with `Submit`, so the channel cannot be closed between the submitter's state check and send. Workers can still read already-queued jobs after closure. Stop message producers before calling this method so no producer continues trying to submit.

#### ⏳ `Shutdown` — drain until a deadline

**Summary:** Stops admission, waits for workers, and cancels jobs if the deadline expires.

**Details:** When `done` closes first, all workers have exited and shutdown returns nil. If the caller context expires first, the pool cancels the job context and returns that context error. Cancellation is cooperative: a job must check its context or use context-aware I/O. On timeout, workers may abandon queued jobs when they observe cancellation, so this RAM-only queue cannot promise those jobs completed.

**What does `ctx.Err()` contain during shutdown?** The `ctx` passed to `Pool.Shutdown` is the drain context. If its deadline expires before the workers finish, `ctx.Err()` returns `context.DeadlineExceeded`; if it is canceled for another reason, it returns `context.Canceled`. If all workers finish before cancellation, `Pool.Shutdown` returns `nil` instead of `ctx.Err()`. The pool uses this error as the shutdown result and cancels the worker job context when the wait times out.

#### 📊 `Stats` — read counters

**Summary:** Returns a concurrency-safe snapshot of the pool's four counters.

**Details:** Atomic loads allow worker updates and reads to happen concurrently. `Accepted` is queue admission, not successful processing; `Completed` and `Failed` are outcomes after a worker runs a job.

#### 👷 `worker` — execute jobs

**Summary:** Repeatedly receives jobs and records each job's result.

**Details:** Each worker exits when the pool context is cancelled or the closed queue is drained. A job error increments `Failed`, then that worker continues to the next job. A nil error increments `Completed`. The worker count therefore caps parallel job execution.

### 🔍 `receiver.go` — file overview

This file routes each validated topic to the right pool and transfers message ownership safely to a worker. Commands and sales use the important-work pool; telemetry uses its own pool so a telemetry burst cannot fill the important queue.

#### 🧩 `Processor` and `Receiver` — declare the route

**Summary:** `Processor` is the business function; `Receiver` holds it and the two pools.

**Details:** The processor receives the worker context and message. The context supports cooperative shutdown. Separate pool fields make the capacity policy visible at the receiver boundary.

#### 📬 `Receiver.Submit` — validate, route, and enqueue

**Summary:** Checks configuration and topic, picks a pool, copies payload bytes, then delegates admission to `Pool.Submit`.

**Details:** A missing processor, invalid topic, or missing pool returns an error before worker execution. Commands and sales select `Important`; other recognized kinds select `Telemetry`. The payload copy prevents the worker from reading bytes that the receive path may reuse or mutate after the method returns. If the message type later adds other mutable slices, maps, or pointers, those need an ownership decision too. The closure passes the pool's worker context to the processor.

### 🔌 Process wiring — connect receive and worker paths

**Summary:** Creates isolated pools, installs the application processor, then submits each received message without doing slow work in the receive loop.

**Details:** `dispatch.New(context.Background(), 2, 16)` configures two workers and sixteen waiting slots for each traffic class. The independent parent context keeps a signal from cancelling workers before the drain; `Pool.Shutdown` cancels jobs after the drain deadline. The processor closure calls the backend handler; in the simulator it calls `respond(...)`. The `case` receives from `inbox`, calls `receiver.Submit` with the signal context, logs errors, and handles non-`Accepted` outcomes. `Full` is an application admission rejection; logging it does not undo an automatic QoS 1 acknowledgement. A command or sales rejection needs an explicit application policy, and durable admission is required if a process crash must not lose accepted jobs.

Save this complete router as `internal/dispatch/receiver.go`. It sends commands, command responses, and sales to the important-work pool, telemetry to its own pool, and rejects unsupported topic kinds before a worker sees them.

```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) })
}
```

Wire the receiver into each process after creating the pools. In the backend, use the Section 5A service as the processor. In the simulator, use its command-response handler as the processor. The receive loop below keeps the callback path short and reports full queues explicitly:

```go
// Backend: keep telemetry and important work from consuming each other's queue slots.
telemetryPool := dispatch.New(context.Background(), 2, 16)
importantPool := dispatch.New(context.Background(), 2, 16)
receiver := &dispatch.Receiver{Telemetry: telemetryPool, Important: importantPool, Process: deviceView.HandleMessage}

case message := <-inbox:
	// Report overload locally; an MQTT acknowledgement cannot undo this loss.
	outcome, err := receiver.Submit(ctx, message)
	if err != nil { log.Printf("message admission error: %v", err); continue }
	if outcome != dispatch.Accepted { log.Printf("message not admitted: outcome=%d topic=%s", outcome, message.Topic) }
```

In the simulator process, construct the same receiver with a processor that runs the existing command handler using the worker pool's context:

```go
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)
	},
}
```

The simulator subscribes to command topics, which the receiver routes to the important-work pool. The backend routes telemetry, sales, and responses through the Section 5A service until later lessons replace its temporary sales/response behavior.

Choose a policy per class: disposable telemetry may be dropped with a counter; commands and sales must produce an explicit admission failure rather than silently disappear. Admission failure under automatic MQTT acknowledgement still loses application work. Record this limitation until durable admission is introduced in Section 13. Do not claim manual acknowledgement alone makes an in-memory queue durable.

When adapting this wiring, keep the pool limits near process startup so they are visible and configurable. Check the returned error before interpreting the outcome: parse/configuration failures are errors, while queue-full and stopped states are outcomes. Decide and record the policy for each message class; the in-memory/automatic-ACK setup cannot guarantee that a rejected command or sales report will be retried.

## 🛑 3. Define ownership and shutdown

> **❓ Question:** **If a worker is still processing when shutdown begins, which components must stay alive until the work finishes or its deadline expires?**

Stop admission before disconnecting MQTT, then let workers finish queued jobs under one shared shutdown deadline. Save this helper as `internal/dispatch/shutdown.go`:

In each complete process listing, the signal case logs and returns from `run`; it does not call the drain itself. The deferred `dispatch.Shutdown` then runs as `run` returns. Since that defer is registered after `defer mqtt.Stop(manager)`, Go's last-in-first-out defer order drains the workers before disconnecting MQTT.

### 📄 `shutdown.go` — file overview

This file coordinates shutdown across all dispatch pools. It stops every pool from accepting work before it waits for any pool, so one queue cannot continue growing while another drains.

```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
}
```

#### ⏳ `Shutdown` — stop and drain all pools

**Summary:** Closes admission on each non-nil pool, then waits for all of them using the same deadline.

**Details:** The first loop calls `StopAdmission` on every pool before the second loop waits. This prevents a later pool from continuing to accept while an earlier one drains. Each pool receives the same context, so the caller's deadline applies to the whole operation rather than restarting per pool. `errors.Join` returns failures from all pool shutdowns, and nil entries are skipped. Stop producers before calling this helper; keep MQTT connected through the drain if workers need it to publish results.

Copy retained byte slices, including future correlation metadata. Protect shared maps with a mutex or give one goroutine ownership. Never hold a state mutex while publishing over the network. Use operation deadlines derived from the application lifecycle.

🧠 **Why these rules matter:** A slice refers to mutable backing storage, so copy bytes that must outlive the callback. Shared maps need a mutex or a single goroutine owner. A network publish can block, so do not hold a state mutex while publishing. Derive operation deadlines from the application lifecycle so workers cannot wait forever on network or storage calls during shutdown.

On shutdown, stop admitting new work and let accepted work finish within a separate bounded drain context. Keep MQTT available during the drain if workers must publish results, then stop the transport producers. Cancel remaining work when the drain deadline expires. Never close a queue while a producer can still send; waiting for worker completion is part of shutdown. Document what happens to unfinished in-memory work.

## 🔗 4. Integrate and stress it

> **❓ Question:** **What observations would show that overload is bounded and shutdown is controlled, rather than merely that the process eventually exits?**

Measure accepted, rejected, completed work, queue capacity, and shutdown time under a finite burst.

Use the receiver wiring above in both processes, with `deviceView.HandleMessage` for the backend and a processor that calls the simulator's `respond(...)` for commands. Keep transport callbacks short, and send messages to the appropriate pool. A finite burst generator should stop after its configured count; do not create one goroutine per incoming message. Record accepted, rejected, completed, and failed counts with the configured worker count and queue capacity.

Apply the same dispatcher boundaries in both programs, but connect each to its own business behavior: the backend's handler and the simulator's ping/pong responder. The finite burst should have a known message count and stop on its own, making accepted/rejected results comparable across runs. Creating a goroutine for every publish would move the unbounded backlog from the channel into goroutine stacks, defeating the pool's memory bound. Include worker count, queue capacity, artificial processing delay, and the four counters in the run record so another person can interpret throughput and overload.

```sh
mkdir -p internal/dispatch journal

gofmt -w internal/dispatch cmd

go test -race ./...

go vet ./...
```

Create `internal/dispatch` and `journal` before saving their files; the `mkdir -p` command is safe to run earlier than the final check block. `go test -race ./...` instruments the test binaries to report unsynchronized memory access during executed tests. It is slower than normal testing and cannot prove unexecuted paths are race-free. `go vet` checks static patterns; it does not replace the race run.

🧰 **What each command does:** `mkdir -p` creates the source and journal directories if needed. `gofmt` formats the Go implementation and command packages. `go test -race ./...` runs the repository tests with race detection for executed paths. `go vet ./...` checks for suspicious Go patterns. Race detection and vet provide different evidence; neither proves paths that were not exercised.

A finite burst generator publishes a configured count and then stops; record its rate, queue capacity, worker count, and artificial delay. A bounded queue has fixed capacity: when workers are slower than arrival, overload becomes measurable instead of consuming memory indefinitely. Worker cancellation is cooperative: handlers must check contexts and use deadlines on blocking operations. Observe accepted/rejected counts and total shutdown time, not just whether the process eventually exits.

The measurements answer different questions: accepted and rejected show admission behavior, completed and failed show worker outcomes, and elapsed shutdown time shows whether the drain respected its bound. A finite generator makes the offered load known; the artificial delay makes the service rate slow enough to trigger pressure. The queue prevents waiting jobs from growing without limit, but it cannot stop one handler from blocking forever unless that handler checks cancellation or uses deadlines for its own network/database calls. Process exit alone is weak evidence because forced termination can discard queued work without reporting it.

To inspect overload, temporarily make a worker slow and publish a finite burst larger than the queue plus active workers. Confirm the measured `Full` outcomes, then reduce the burst and confirm normal handling resumes. Observe shutdown time under the deadline; do not infer that queued work completed solely because the process exited.

The slow-worker run should push arrivals beyond the pool's total immediate capacity: active worker slots plus waiting queue slots. `Full` outcomes confirm that backpressure is bounded and visible. A smaller burst should fit under available capacity or be drained quickly enough to avoid rejection, showing that the system recovers after pressure rather than remaining stuck. During shutdown, compare elapsed time to the configured deadline and inspect completed/failed counts; jobs still queued or cancelled at expiry were not proven complete. Then compare that application evidence with the MQTT acknowledgement trace from the first section.

**Done when:** A burst causes observable bounded overload, normal ping/pong recovers afterward, shutdown finishes by its deadline, and the race detector passes. Explain the difference between broker acknowledgement, queue admission, and completed application work.

## ✅ Result — the story so far

The receive callback now has a defined small job: copy and admit work into bounded queues, then return. Workers validate and handle admitted messages, while counters make overload visible and shutdown follows a deliberate stop-admission, drain, and disconnect order. In this lesson's in-memory design, automatic MQTT acknowledgement can happen before worker completion, so a process crash can still lose queued work; durable admission is a later step.

Previous: [05A — Backend message processing](05a-backend-message-processing.md). Next: [07-command-response.md](07-command-response.md).