mqtt-5-learning/lessons/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 before handling them. 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.

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.

### ↩️ 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.

### 🧪 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.

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 signal context in the early example cancels AutoPaho immediately, so a real drain policy needs a separately controlled transport lifetime.

<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.

## 🛠️ 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.

Save this complete file as `internal/dispatch/pool.go`. It provides bounded nonblocking admission, explicit outcomes, worker accounting, and a shutdown deadline.

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

The complete wiring above creates separate pools for traffic classes that must not consume one another's capacity. It submits work from the receive loop and reports `Full` as an explicit application rejection; under automatic MQTT acknowledgement that rejection does not undo the acknowledgement.

Save this complete router as `internal/dispatch/receiver.go`. It sends commands 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.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. The receive loop below keeps the callback path short and reports full queues explicitly:

```go
// Keep telemetry and important work from consuming each other's queue slots.
telemetryPool := dispatch.New(ctx, 2, 16)
importantPool := dispatch.New(ctx, 2, 16)
receiver := &dispatch.Receiver{Telemetry: telemetryPool, Important: importantPool, Process: func(jobCtx context.Context, message mqtt.Message) error {
	return handleMessage(message) // Use respond(...) in the simulator.
}}

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

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.

## 🛑 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`:

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

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.

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. Keep transport callbacks short, and send validated 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.

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

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.

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.

**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: [05-contracts-validation.md](05-contracts-validation.md). Next: [07-command-response.md](07-command-response.md).