From f9ab91ed60d7dcbe6a436754045c2484891bc396 Mon Sep 17 00:00:00 2001 From: ryan Date: Tue, 13 Oct 2026 18:08:17 +0330 Subject: [PATCH] feat: expand MQTT contract and reliability lessons --- COURSE.md | 43 ++ internal/contracts/decode.go | 36 ++ internal/contracts/decode_test.go | 76 ++++ internal/contracts/message.go | 180 +++++++++ internal/contracts/topics.go | 1 + lab/mosquitto/acl | 2 + lessons/00-system-lifecycle.md | 61 ++- lessons/05-contracts-validation.md | 517 +++++++++++++++++++++++-- lessons/06-concurrency-cancellation.md | 257 +++++++++++- lessons/07-command-response.md | 485 ++++++++++++++++++++++- lessons/08-http-commands.md | 284 +++++++++++++- lessons/09-sessions-recovery.md | 14 +- lessons/10-expiry-presence-state.md | 14 +- lessons/11-security-boundaries.md | 10 +- lessons/11b-dynamic-security.md | 166 ++++++++ lessons/16-mqtt5-controls.md | 24 ++ lessons/README.md | 4 +- 17 files changed, 2083 insertions(+), 91 deletions(-) create mode 100644 internal/contracts/decode.go create mode 100644 internal/contracts/decode_test.go create mode 100644 internal/contracts/message.go create mode 100644 internal/contracts/topics.go create mode 100644 lessons/11b-dynamic-security.md diff --git a/COURSE.md b/COURSE.md index e70c909..cea5ce4 100644 --- a/COURSE.md +++ b/COURSE.md @@ -221,6 +221,19 @@ Detailed lesson: [Section 11 explanations and experiments](lessons/11-security-b **Done when:** Positive and negative access tests behave as designed. Record the observed reason/error for denied operations; do not assume every broker denial is reported identically. +### Optional Section 11B β€” Provision device ACLs with Mosquitto Dynamic Security + +Detailed lesson: [Section 11B Dynamic Security implementation](lessons/11b-dynamic-security.md). + +Use this extension when you want to replace the lab's static password/ACL files with live per-device provisioning. It uses a separate broker configuration and state file so the working Section 1 lab remains intact. + +1. [ ] Enable the plugin in an isolated Mosquitto configuration and persist its state file with broker-readable permissions. +2. [ ] Create least-privilege device/backend roles, using a trusted identity-to-topic rule and default-deny behavior. +3. [ ] Provision one unique device client and assign its role through a separately protected management identity. +4. [ ] Test allowed and denied publish/subscribe operations, mismatched payload identity, revocation, and state persistence after broker restart. + +**Done when:** You can trace provisioning, broker-side ACL enforcement, backend validation, and revocation of a connected device. The production runtime should not receive broker-administrator credentials merely to process messages. + ## Section 12 β€” Implement representative vending workflows **Outcome:** You understand the backend's main device-facing jobs using synthetic data. @@ -237,6 +250,36 @@ Detailed lesson: [Section 11 explanations and experiments](lessons/11-security-b **Outcome:** Important state and deduplication survive application restart. +### πŸ“₯ Choose the acknowledgement boundary for important work + +The MQTT client receives the broker's QoS 1 PUBLISH and invokes the registered callback. In automatic acknowledgement mode, Paho sends PUBACK after the callback returnsβ€”even if the callback returns an error. Returning an error does not ask the broker to retry. A callback that places a message in a RAM queue and returns can therefore acknowledge it before a worker handles it; a process crash can then lose work the broker considers delivered. + +With Paho's manual acknowledgement enabled, the application calls `received.Client.Ack(received.Packet)` when its chosen acceptance boundary has been reached. For important reports, make that boundary a successful database transaction: validate the topic and payload, then atomically save the message/deduplication key and the business effect, and only then acknowledge. Keep the original client and packet available to the worker until `Ack` succeeds. Manual acknowledgement delays protocol completion; it does not make an in-memory queue durable. + +> **❓ Question:** **If the database is unavailable, should an important report be acknowledged?** +> +> **πŸ’¬ Answer:** Not before the application has safely recorded it. Leave it unacknowledged and follow a deliberate connection/recovery policy, or reject/quarantine it durably if it is permanently invalid. Returning a callback error alone neither acknowledges nor requests redelivery in manual mode. Avoid leaving poison messages unacknowledged forever: they consume in-flight capacity and can stall later QoS 1/2 deliveries. + +### πŸ” Make database effects idempotent + +QoS 1 can deliver a message again when a connection fails before the broker receives PUBACK. Application retries and crash recovery can also repeat work. Use a stable application-level identityβ€”such as `(device_id, report_id)`β€”and enforce it with a database unique constraint. In one transaction, decide whether this is new work, record the deduplication key, and apply the business effect. A duplicate should return or reproduce the prior application result without applying the effect twice. MQTT packet identifiers are temporary protocol identifiers, so they are not suitable business deduplication keys. + +For sales reports, commit the report record and its inventory/revenue effect atomically. For command results, commit the validated result and state transition together. A stable ID helps the application recognize repeats; the database constraint makes that decision safe across concurrent workers and process restarts. + +### 🧹 Give invalid messages a durable disposition + +Topic parsing and payload validation happen before business state changes. If a message is permanently invalid, record a bounded rejection reason and enough safe identity/context to diagnose it, or move it to a durable quarantine. After that disposition commits, acknowledge it so it does not block the stream. Do not log credentials or dump untrusted payloads. If failure is transientβ€”for example, the database is temporarily unavailableβ€”do not mark the message permanently rejected; keep it recoverable and apply a bounded retry/reconnect policy. + +### βš–οΈ Match QoS and queue policy to the data + +QoS 1 is a practical choice for important ingestion when the application is built to tolerate duplicates. QoS 2 adds a protocol handshake; it does not make a database transaction or physical device effect exactly once. For disposable telemetry, automatic acknowledgement and bounded best-effort processing can be acceptable when drops are counted and freshness matters more than replay. For reports and command state, prefer durable admission and idempotent processing. + +Keep the callback short: copy the needed message data and submit it to a bounded worker queue. For important work, preserve the acknowledgement handle with the queued item and acknowledge only after the durable transaction commits. Bound queue capacity and configure MQTT 5 Receive Maximum; monitor queue depth, unacknowledged deliveries, processing failures, duplicate counts, and acknowledgement latency. Receive Maximum limits protocol in-flight QoS 1/2 deliveries, not the size of the Go queue. + +If the transaction must also trigger a later MQTT publication, do not commit the database and then rely on an immediate publish alone: a crash between the commit and publication can lose the response. Store an outbox entry in the same transaction and publish it from a retrying worker. Section 14 develops that crash-recovery pattern. + +πŸ“š **Builds on:** [Lesson 5](lessons/05-contracts-validation.md) validates topics and payloads; [Lesson 6](lessons/06-concurrency-cancellation.md) explains callbacks, manual acknowledgement, and bounded queues; [Lesson 7](lessons/07-command-response.md) introduces stable IDs and idempotent effects; [Lesson 16](lessons/16-mqtt5-controls.md) covers Receive Maximum. Section 14 adds the transactional outbox. + 1. [ ] Add a dedicated lab PostgreSQL instance and migrations for devices, commands, command results, and sales reports. Use `database/sql` with pgx stdlib to match the observed connector style. 2. [ ] Move important in-memory repositories behind small interfaces. Use contexts, query deadlines, bounded connection pools, and transaction boundaries. 3. [ ] Enforce uniqueness for device/report IDs and command IDs. Persist a deduplication decision and its corresponding database effect atomically. diff --git a/internal/contracts/decode.go b/internal/contracts/decode.go new file mode 100644 index 0000000..265bb17 --- /dev/null +++ b/internal/contracts/decode.go @@ -0,0 +1,36 @@ +package contracts + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "time" +) + +const MaxPayloadBytes = 16 * 1024 + +// Decode bounds input, rejects unknown fields, requires exactly one JSON value, +// and validates the decoded MQTT contract against its parsed topic. +func Decode[T MQTTMessage](payload []byte, topic Topic, now time.Time) (T, error) { + var value T + if len(payload) > MaxPayloadBytes { + return value, fmt.Errorf("payload exceeds %d bytes", MaxPayloadBytes) + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&value); err != nil { + return value, fmt.Errorf("decode JSON: %w", err) + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return value, fmt.Errorf("payload contains more than one JSON value") + } + return value, fmt.Errorf("trailing data after JSON value: %w", err) + } + if err := value.Validate(topic, now); err != nil { + return value, fmt.Errorf("validate message: %w", err) + } + return value, nil +} diff --git a/internal/contracts/decode_test.go b/internal/contracts/decode_test.go new file mode 100644 index 0000000..7bb4247 --- /dev/null +++ b/internal/contracts/decode_test.go @@ -0,0 +1,76 @@ +package contracts + +import ( + "strings" + "testing" + "time" +) + +var testTopic = Topic{DeviceID: "device-001", Kind: KindTelemetry} +var testNow = time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + +// Topic tests reject malformed routes and unsupported message kinds. +func TestParseTopic(t *testing.T) { + topic, err := ParseTopic("lab/v1/devices/device-001/sales") + if err != nil { + t.Fatal(err) + } + if topic.DeviceID != "device-001" || topic.Kind != KindSales { + t.Fatalf("unexpected topic: %#v", topic) + } + for _, invalid := range []string{"lab/v1/devices/+/sales", "lab/v1/devices/device-001/unknown", "lab/v1/devices/device-001/sales/extra"} { + if _, err := ParseTopic(invalid); err == nil { + t.Errorf("ParseTopic(%q) succeeded", invalid) + } + } +} + +// These cases cover strict JSON shape and single-value decoding. +func TestDecodeRejectsUnknownAndTrailingValues(t *testing.T) { + if _, err := Decode[Heartbeat]([]byte(`{"schema_version":1,"unknown":true}`), testTopic, testNow); err == nil { + t.Fatal("unknown field was accepted") + } + if _, err := Decode[Heartbeat]([]byte(`{"schema_version":1,"message_id":"m-1","device_id":"device-001","occurred_at":"2026-10-13T12:00:00Z"} {"message_id":"m-2"}`), testTopic, testNow); err == nil { + t.Fatal("trailing JSON value was accepted") + } +} + +// This case ensures the application limit is applied before decoding. +func TestDecodeRejectsOversizedPayload(t *testing.T) { + payload := []byte(`"` + strings.Repeat("x", MaxPayloadBytes) + `"`) + if _, err := Decode[Heartbeat](payload, testTopic, testNow); err == nil { + t.Fatal("oversized payload was accepted") + } +} + +// Decode validates the message against both its parsed route and supplied clock. +func TestDecodeValidatesContract(t *testing.T) { + payload := []byte(`{"schema_version":1,"message_id":"m-1","device_id":"device-002","occurred_at":"2026-10-13T12:00:00Z"}`) + if _, err := Decode[Heartbeat](payload, testTopic, testNow); err == nil { + t.Fatal("topic/body identity mismatch was accepted") + } +} + +// The type parameter must implement MQTTMessage; successful decoding also validates it. +func TestDecodeAcceptsOnlyValidatedMQTTMessages(t *testing.T) { + payload := []byte(`{"schema_version":1,"message_id":"m-1","device_id":"device-001","occurred_at":"2026-10-13T12:00:00Z"}`) + message, err := Decode[Heartbeat](payload, testTopic, testNow) + if err != nil { + t.Fatal(err) + } + if message.DeviceID != testTopic.DeviceID { + t.Fatalf("device_id = %q, want %q", message.DeviceID, testTopic.DeviceID) + } +} + +// A body cannot override the device selected by the topic. +func TestCommandRejectsTopicIdentityMismatch(t *testing.T) { + topic, err := ParseTopic("lab/v1/devices/device-001/commands") + if err != nil { + t.Fatal(err) + } + command := Command{SchemaVersion: SchemaVersion, MessageID: "m-1", DeviceID: "device-002", OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: "c-1", Command: "ping"} + if err := command.Validate(topic, time.Now().UTC()); err == nil { + t.Fatal("topic/body identity mismatch was accepted") + } +} diff --git a/internal/contracts/message.go b/internal/contracts/message.go new file mode 100644 index 0000000..be6bf83 --- /dev/null +++ b/internal/contracts/message.go @@ -0,0 +1,180 @@ +package contracts + +import ( + "fmt" + "strings" + "time" +) + +const SchemaVersion = 1 + +// Kind values match the final topic segment used to route each message. +type Kind string + +const ( + KindTelemetry Kind = "telemetry" + KindCommands Kind = "commands" + KindResponses Kind = "responses" + KindSales Kind = "sales" +) + +// Topic is the parsed device scope and message kind from an MQTT topic. +type Topic struct { + DeviceID string + Kind Kind +} + +// MQTTMessage is implemented by each versioned message contract. +// Validation uses the parsed route and caller-supplied time for deterministic checks. +type MQTTMessage interface { + Validate(topic Topic, now time.Time) error +} + +// These types define the JSON fields shared by backend and simulator. +type Heartbeat struct { + SchemaVersion int `json:"schema_version"` + MessageID string `json:"message_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + Temperature *float64 `json:"temperature,omitempty"` +} + +type Command struct { + SchemaVersion int `json:"schema_version"` + MessageID string `json:"message_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + CommandID string `json:"command_id"` + Command string `json:"command"` +} + +type CommandResult struct { + SchemaVersion int `json:"schema_version"` + MessageID string `json:"message_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + CommandID string `json:"command_id"` + Status string `json:"status"` + Detail string `json:"detail,omitempty"` +} + +type SalesLine struct { + SKU string `json:"sku"` + Quantity int64 `json:"quantity"` + AmountMinor int64 `json:"amount_minor"` +} + +type SalesReport struct { + SchemaVersion int `json:"schema_version"` + ReportID string `json:"report_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + Items []SalesLine `json:"items"` +} + +// validID limits identifiers to safe, single-segment topic values. +func validID(value string) bool { + if value == "" || len(value) > 128 { + return false + } + for _, r := range value { + if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._-", r)) { + return false + } + } + return true +} + +// validateEnvelope checks fields shared by every versioned message. +func validateEnvelope(topic Topic, version int, messageID, deviceID, occurredAt string, now time.Time) error { + if version != SchemaVersion { + return fmt.Errorf("unsupported schema_version %d", version) + } + if !validID(messageID) || !validID(deviceID) { + return fmt.Errorf("message_id and device_id must be valid IDs") + } + if topic.DeviceID != deviceID { + return fmt.Errorf("topic device %q does not match payload device_id %q", topic.DeviceID, deviceID) + } + stamp, err := time.Parse(time.RFC3339, occurredAt) + if err != nil { + return fmt.Errorf("occurred_at must be RFC3339: %w", err) + } + if stamp.After(now.Add(30 * time.Second)) { + return fmt.Errorf("occurred_at is more than 30 seconds in the future") + } + return nil +} + +// Each Validate method adds checks specific to one message kind. +func (m Heartbeat) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindTelemetry { + return fmt.Errorf("heartbeat requires telemetry topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if m.Temperature != nil && (*m.Temperature < -50 || *m.Temperature > 100) { + return fmt.Errorf("temperature is outside the supported range") + } + return nil +} + +func (m Command) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindCommands { + return fmt.Errorf("command requires commands topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if !validID(m.CommandID) || m.Command != "ping" { + return fmt.Errorf("command_id is invalid or command is unsupported") + } + return nil +} + +func (m CommandResult) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindResponses { + return fmt.Errorf("command result requires responses topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if !validID(m.CommandID) || (m.Status != "accepted" && m.Status != "completed" && m.Status != "rejected") { + return fmt.Errorf("command result has invalid command_id or status") + } + return nil +} + +func (m SalesReport) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindSales { + return fmt.Errorf("sales report requires sales topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.ReportID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if len(m.Items) == 0 || len(m.Items) > 100 { + return fmt.Errorf("sales report must contain 1 to 100 items") + } + for i, item := range m.Items { + if !validID(item.SKU) || item.Quantity <= 0 || item.AmountMinor < 0 { + return fmt.Errorf("item %d has invalid SKU, quantity, or amount_minor", i) + } + } + return nil +} + +// ParseTopic turns broker routing metadata into values handlers can compare. +func ParseTopic(value string) (Topic, error) { + parts := strings.Split(value, "/") + if len(parts) != 5 || parts[0] != "lab" || parts[1] != "v1" || parts[2] != "devices" || !validID(parts[3]) { + return Topic{}, fmt.Errorf("topic must be lab/v1/devices/{deviceID}/{kind}") + } + kind := Kind(parts[4]) + switch kind { + case KindTelemetry, KindCommands, KindResponses, KindSales: + return Topic{DeviceID: parts[3], Kind: kind}, nil + default: + return Topic{}, fmt.Errorf("unsupported topic kind %q", kind) + } +} diff --git a/internal/contracts/topics.go b/internal/contracts/topics.go new file mode 100644 index 0000000..6cdb93a --- /dev/null +++ b/internal/contracts/topics.go @@ -0,0 +1 @@ +package contracts diff --git a/lab/mosquitto/acl b/lab/mosquitto/acl index 08f5527..c5abd02 100644 --- a/lab/mosquitto/acl +++ b/lab/mosquitto/acl @@ -2,8 +2,10 @@ user backend topic write lab/v1/devices/+/commands topic read lab/v1/devices/+/telemetry topic read lab/v1/devices/+/responses +topic read lab/v1/devices/+/sales user simulator topic read lab/v1/devices/device-001/commands topic write lab/v1/devices/device-001/telemetry topic write lab/v1/devices/device-001/responses +topic write lab/v1/devices/device-001/sales diff --git a/lessons/00-system-lifecycle.md b/lessons/00-system-lifecycle.md index b34c0a2..b6ca934 100644 --- a/lessons/00-system-lifecycle.md +++ b/lessons/00-system-lifecycle.md @@ -34,7 +34,9 @@ The broker has its own setup and runtime. Each client has its own configuration ## πŸ–₯️ 1. Set up the broker -> πŸ’‘ **Story question:** Before a client arrives, how does the broker know where to listen, which credentials to accept, and which topics each identity may use? +**❓ Question:** **Before a client arrives, how does the broker know where to listen, which credentials to accept, and which topics each identity may use?** + +**πŸ’¬ Answer:** Broker configuration supplies the listener, authentication source, and topic permissions. The clients present credentials; they do not create their own broker accounts. Those decisions belong to broker configuration or to an authentication/authorization service configured by the broker. In this lab, Mosquitto is configured with a listener, a password file, and an ACL file. The `backend` and `simulator` accounts are created before the clients connect. @@ -58,7 +60,9 @@ For this learning environment, the broker is isolated and uses synthetic lab dat ## βš™οΈ 2. Configure the backend and device -> πŸ’‘ **Story question:** If both programs are MQTT clients, what distinguishes the backend's connection from the device's connection? +**❓ Question:** **If both programs are MQTT clients, what distinguishes the backend's connection from the device's connection?** + +**πŸ’¬ Answer:** Each process has its own Client ID, credentials, configuration, and subscriptions. In this lab, the username selects broker ACL rules; a Client ID identifies an MQTT connection/session and is not a password. Each process has its own configuration and client identity. Both need a broker URL and credentials accepted by that broker. They also need distinct Client IDs while running at the same time. Their usernames determine which broker ACL rules apply in this lab; Client ID identifies the MQTT connection/session and is not a password. @@ -82,7 +86,9 @@ Connection settings are distinct from message choices. Keep Alive, session behav ## πŸ”Œ 3. Connect both clients -> πŸ’‘ **Story question:** Once the Go process has configuration, what must it prove before it can claim to be connected? +**❓ Question:** **Once the Go process has configuration, what must it prove before it can claim to be connected?** + +**πŸ’¬ Answer:** It needs an accepted CONNACK. Creating a connection manager is not enough, and a successful connection does not prove that subscriptions or business operations succeeded. The program parses the broker URL, creates a Paho/AutoPaho configuration, and starts the connection manager. AutoPaho manages the connection lifecycle and reconnect attempts; Paho implements MQTT protocol exchanges. The manager being created is not yet proof of a successful connection. The broker must accept MQTT `CONNECT` with `CONNACK`, and the application can wait for that state with `AwaitConnection`. @@ -107,7 +113,9 @@ The backend and device go through this independently. A successful `CONNACK` pro ## πŸ“¬ 4. Register subscriptions -> πŸ’‘ **Story question:** The broker accepted the connection. How does it learn which messages this client wants to receive? +**❓ Question:** **The broker accepted the connection. How does it learn which messages this client wants to receive?** + +**πŸ’¬ Answer:** The client sends SUBSCRIBE topic filters, and the broker reports each filter's result in SUBACK. The local Paho handler is a separate step that processes publications the broker delivers. The client sends `SUBSCRIBE` requests containing topic filters. The broker checks each request against its rules and returns `SUBACK` results. In the Go application, a local callback is also configured so Paho knows what to do when a matching publication arrives. The broker-side subscription and the Go-side callback are two separate parts of receiving a message. @@ -130,7 +138,9 @@ The backend commonly listens for telemetry and responses. The device listens for ## πŸ“¨ 5. Publish and route messages -> πŸ’‘ **Story question:** Once subscriptions exist, how does a command travel from backend code to the device application? +**❓ Question:** **Once subscriptions exist, how does a command travel from backend code to the device application?** + +**πŸ’¬ Answer:** The backend publishes bytes to a topic, the broker routes them to matching subscriptions, and the device's MQTT client passes the publication to application code. QoS acknowledgements cover a delivery hop; the device's application response carries business evidence. The backend serializes an application value into bytes, publishes those bytes to a concrete topic, and the broker routes the publication to matching subscribed clients. Paho's receive callback runs inside the client process; the application can copy and enqueue the message for its main loop to handle. @@ -157,7 +167,9 @@ There are separate MQTT delivery hops. A `PUBACK` on the backend's publication c ## πŸ”„ 6. Recover and stop -> πŸ’‘ **Story question:** What should happen if the broker or network disappears while the backend process remains alive? +**❓ Question:** **What should happen if the broker or network disappears while the backend process remains alive?** + +**πŸ’¬ Answer:** The client can reconnect, but the application must ensure needed subscriptions are present. Offline delivery depends on session settings, QoS, expiry, and broker state, so reconnect alone does not guarantee that every message is recovered. AutoPaho can reconnect according to its lifecycle policy. After a connection comes back, the application must ensure required subscriptions are presentβ€”either because the broker resumed a session or because the client re-subscribed. In this lab, Lesson 4's helper explicitly subscribes again on connection-up. Whether offline publications arrive depends on MQTT session settings, QoS, broker state, and expiry; reconnect alone does not promise recovery of every message. @@ -185,35 +197,50 @@ stateDiagram-v2 | Lifecycle area | Lessons | Evidence to understand | |---|---|---| -| Broker setup, credentials, and topic permissions | [1](01-lab-setup.md), [11](11-security-boundaries.md) | Which process owns each setting; authentication versus authorization | +| Broker setup, credentials, and topic permissions | [1](01-lab-setup.md), [11](11-security-boundaries.md), [11B](11b-dynamic-security.md, optional) | Mosquitto owns broker authentication and topic ACL enforcement; optional live per-device provisioning | | Go client configuration | [2](02-go-connection.md), [3](03-typed-configuration.md) | Effective settings, identity, validation, and precedence | | CONNECT, CONNACK, reconnect, and shutdown | [2](02-go-connection.md), [4](04-publish-subscribe-simulator.md), [9](09-sessions-recovery.md) | What connection acceptance proves and what must be restored | -| Subscriptions and broker routing | [4](04-publish-subscribe-simulator.md), [9](09-sessions-recovery.md), [16](16-mqtt5-controls.md) | SUBACK results, matching filters, session restoration, and options | -| Publish, delivery, and application response | [4](04-publish-subscribe-simulator.md), [5](05-contracts-validation.md) | Per-hop QoS acknowledgements versus validated business evidence | +| Topic names, filters, wildcards, and routing | [1](01-lab-setup.md), [4](04-publish-subscribe-simulator.md), [16](16-mqtt5-controls.md) | Concrete topic names, `+`/`#` filter matching, broker routing, and subscription options | +| Subscriptions and broker routing | [4](04-publish-subscribe-simulator.md), [9](09-sessions-recovery.md), [16](16-mqtt5-controls.md) | SUBACK results, matching filters, session restoration, shared subscriptions, and options | +| QoS 0/1/2, duplicates, and acknowledgement scope | [4](04-publish-subscribe-simulator.md), [6](06-concurrency-cancellation.md), [7](07-command-response.md), [16](16-mqtt5-controls.md) | Per-hop packet evidence, DUP/retries, and why QoS does not make business effects exactly once | +| Topic and payload validation in Go | [5](05-contracts-validation.md) | `ParseTopic` checks topic shape and kind; `Decode` validates JSON, schema, route/body device identity, and message values before handling | | Callback work, queues, and overload | [6](06-concurrency-cancellation.md) | Admission, acknowledgement timing, bounded capacity, and shutdown | | Reliable command/response and HTTP | [7](07-command-response.md), [8](08-http-commands.md) | Correlation, idempotency, timeout, and status reporting | -| Expiry and device presence | [10](10-expiry-presence-state.md) | Stale messages and the limits of retained state | +| Sessions, retained state, expiry, and presence | [9](09-sessions-recovery.md), [10](10-expiry-presence-state.md) | Offline delivery prerequisites, retained desired state, message expiry, Will, and heartbeat freshness | +| MQTT 5 flow controls and capabilities | [16](16-mqtt5-controls.md) | Receive Maximum, packet limits, reason codes, aliases, and capability boundaries | -The table links each part of this lifecycle to the lessons that explain or exercise it. Sections 12–15 and 17–18 in [the course plan](../COURSE.md) extend the story into persistence, process boundaries, and operations. +The table points to the lessons that explain and exercise each part of the lifecycle. Lessons 1–11 and 16 contain detailed instructions; Sections 12–15 and 17–18 in [the course plan](../COURSE.md) are currently guided checklists that extend the story into vending workflows, persistence, process boundaries, and operations. + +### What validates a topic? + +Lesson 5's `contracts.ParseTopic` and each message's `Validate` method are necessary backend checks. They reject malformed routes, unsupported message kinds, and a payload whose claimed `device_id` does not match the topic. The typed decoder then checks the payload contract before application handling. + +Those checks do not authorize the publisher. A client with valid credentials could still try to publish to another device's topic, so Mosquitto must enforce a least-privilege ACL for both publish and subscribe operations. Lesson 11 tests that broker boundary and the separate HTTP-user authorization boundary; optional Lesson 11B covers per-device ACL provisioning. Authentication identifies a broker client, ACLs authorize topic operations, and Lesson 5 validates message content. Use all three boundaries. --- ## πŸ§ͺ Understanding check 1. What does a successful `CONNACK` prove, and what does it not prove? -2. Why does the backend need both a broker subscription and a local receive handler? -3. Which acknowledgement tells the backend that the device application completed a command? -4. What is the difference between a client credential and a Client ID in this lab? +2. How are a topic name, a subscription filter, and an ACL rule different? +3. Why does the backend need both a broker subscription and a local receive handler? +4. Which acknowledgement tells the backend that the device application completed a command? +5. What is the difference between backend topic parsing/validation and broker ACL authorization? +6. What is the difference between a client credential and a Client ID in this lab? > [!answer]- πŸ“‹ Answers > > **1.** It proves the broker accepted that MQTT connection. It does not prove a subscription was accepted, a device is online, or business work completed. > -> **2.** The broker subscription controls routing; the local handler controls what Paho/the application does with a delivered PUBLISH. +> **2.** A topic name labels one publication route. A filter can match multiple topic names when the broker routes to a subscription; `+` matches one level and `#` matches the remaining levels. An ACL rule permits or denies a client's publish or subscribe operation. > -> **3.** None of the MQTT packet acknowledgements proves application completion. The device must publish an application-level response, and the backend must validate it. +> **3.** The broker subscription controls routing; the local handler controls what Paho/the application does with a delivered PUBLISH. > -> **4.** Credentials authenticate a client and select its ACL identity here. Client ID identifies the MQTT connection/session and can be configured as an ACL input, but it is not inherently authentication. +> **4.** None of the MQTT packet acknowledgements proves application completion. The device must publish an application-level response, and the backend must validate it. +> +> **5.** Backend validation checks message shape and consistency after delivery. Broker ACLs decide whether that client may publish or subscribe to the topic at all. Payload validation does not replace authorization. +> +> **6.** Credentials authenticate a client and select its ACL identity here. Client ID identifies the MQTT connection/session and can be configured as an ACL input, but it is not inherently authentication. --- diff --git a/lessons/05-contracts-validation.md b/lessons/05-contracts-validation.md index 01a3bdb..e14f53e 100644 --- a/lessons/05-contracts-validation.md +++ b/lessons/05-contracts-validation.md @@ -2,9 +2,9 @@ ## 🎬 Start with the system story -> πŸ’‘ **Story question:** The broker can route a message because its topic matches. How does the Go backend decide that the topic and payload describe a valid message from the expected device? +> **❓ Question:** **The broker can route a message because its topic matches. How does the Go backend decide that the topic and payload describe a valid message from the expected device?** -Routing is not validation or identity proof. A message arrives as a topic plus bytes; the application must parse the route, decode the payload, and check that fields and identity make sense before changing state. A device ID inside JSON is a claim, so broker authentication and topic authorization remain separate safeguards. +**πŸ’¬ Answer:** The topic provides the route, but it does not verify the payload. The backend parses the topic, validates the JSON, and compares the claimed device ID before business handling. Broker authentication and ACLs separately control who may publish to a topic. ```mermaid flowchart LR @@ -20,20 +20,24 @@ flowchart LR Prerequisite: Section 4 exchanges heartbeats and ping/pong. The clients share the disposable password in `configs/lab.toml`; select the simulator account with `-username simulator` when starting it. -Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. Create files at the project-relative paths shown below, not under `lessons/`. Continue the same application and pinned Paho v0.23.0 dependency. These are guided implementation exercises; attempt each step before requesting a full solution. +Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. Save the complete code blocks at their project-relative paths, not under `lessons/`. This lesson provides a full working implementation with the pinned Paho v0.23.0 dependency; use the exercises to inspect and verify its behavior. ## πŸ“¨ Follow one message through the trust boundaries -Suppose the backend receives topic `lab/v1/devices/device-001/sales` with a body claiming `device_id: device-002`. The broker may have correctly authorized the sender to publish to the topic. It has not established that the JSON is truthful. Reject the mismatch before selecting a device record or changing a balance. +**❓ Question:** **The backend receives a sales report on device-001's topic, but its JSON says `device_id: device-002`. What could go wrong if the backend accepts it?** + +**πŸ’¬ Answer:** The report could be applied to the wrong device. The broker may allow the publish, but it does not compare the topic with the JSON. The backend compares both IDs and rejects a mismatch before changing data. ```text PUBLISH β†’ size limit β†’ topic parser β†’ JSON decoder β†’ schema checks β†’ topic/body identity check β†’ business validation β†’ handler ``` -Each step answers a different question. The size limit bounds decoding cost. Parsing establishes the intended route. JSON decoding proves structure only. Business validation rejects a syntactically valid negative quantity. None of these replaces broker authentication. +The backend applies several checks because each catches a different problem. A size limit bounds the payload it decodes. The topic parser identifies the device and message type. JSON decoding checks the data's shape, while business validation checks its values. For example, a sales report can be valid JSON but contain a negative quantity. The backend rejects that value. These checks do not replace broker authentication or ACLs. -The 16 KiB application limit applies to `Payload`, not the entire MQTT packet. Topic and properties add bytes; the library has already received the packet before this validator runs. Lesson 16 adds protocol packet-size limits to bound an earlier layer. +**❓ Question:** **Does the 16 KiB application limit cap the whole MQTT packet?** + +**πŸ’¬ Answer:** No. It caps the payload the application decodes. The topic and properties add packet bytes, and Paho has already received the packet before this check runs. Lesson 16 covers an earlier packet-size limit. ### πŸ“¨ Map encoding metadata into Paho @@ -51,9 +55,11 @@ properties := &paho.PublishProperties{ _ = properties ``` -Payload Format Indicator 1 declares UTF-8 character data. It does not mean JSON, schema version 1, or trusted content. Content Type describes the media type. For example, `hello` is UTF-8 but fails your JSON object contract. Property pointers distinguish an omitted property from an explicitly supplied value; inspect nil before dereferencing incoming fields. +**❓ Question:** **A message declares JSON, but its payload is `hello`. Is it valid JSON?** -Keep these distinctions in your transport type when needed: packet metadata belongs alongside payload bytes, while `schema_version` and `command_id` belong to your application contract. Do not make handlers depend on a mutable packet owned elsewhere merely to read metadata. +**πŸ’¬ Answer:** No. Content Type describes the media type; Payload Format Indicator 1 says the bytes are UTF-8 text. Neither validates the content or proves its schema. Decode the bytes, and check an incoming property pointer for nil before reading it; nil means that property was omitted. + +Keep MQTT properties with the received transport message when the application needs them. Fields such as `schema_version` and `command_id` belong in the JSON contract type. This lets handlers use the needed data without depending on a packet object that another part of the program may change. ### πŸ§ͺ Predict, modify, explain @@ -62,49 +68,508 @@ Send a valid heartbeat, then change only its body device ID, then change only it
Why can an invalid message still get a successful MQTT acknowledgement? -MQTT receipt and application validity are separate. With the early automatic-acknowledgement helper, returning from the callback allows acknowledgement even when later validation rejects the queued message. Record rejection in application evidence; do not report the PUBACK as successful business processing. Only respond to a command after establishing a safe, authorized response destination. +**❓ Question:** **The broker delivers a sales report with quantity `-3`. Can the MQTT client acknowledge it even if the application rejects it?** + +**πŸ’¬ Answer:** Yes. In this helper, the callback can return and the MQTT client may acknowledge delivery. The backend then records the validation failure and rejects the sale. PUBACK confirms the MQTT delivery step; it does not mean the sale succeeded.
## πŸ“‹ 1. Create the contract package -> πŸ’‘ **Story question:** What shared vocabulary lets the backend and device agree on the shape and meaning of each message? +> **❓ Question:** **What shared vocabulary lets the backend and device agree on the shape and meaning of each message?** -The contract package defines those wire-level names and fields so neither client silently invents a different JSON shape. +The contract package gives both programs the same field names and meanings. For example, if the simulator sends `device` while the backend expects `device_id`, the message may be rejected or misunderstood. ```sh mkdir -p internal/contracts ``` -Create `internal/contracts/messages.go`, `topics.go`, `decode.go`, and `decode_test.go`. Edit the existing backend and simulator files to use this package. These are exercises: implement the functions described below; this lesson does not supply a complete replacement program. +The complete implementation for this step is shown below. Save each block at its stated path, then update the backend and simulator to use these contracts before they handle or publish messages. -Define JSON structs for Heartbeat, Command, CommandResult, and SalesReport. Use `schema_version`, `message_id`, `device_id`, and `occurred_at` consistently. Command also needs `command_id` and `command`; CommandResult needs the same command ID and a status. A sales report needs a stable report ID and synthetic line items with positive quantities and integer minor-unit amounts. Keep command identity stable across retries; an MQTT packet identifier is not that identity. +Save the first block as `internal/contracts/messages.go`. It defines the versioned messages, topic parser, and value checks. A command ID remains stable across retries; an MQTT packet identifier identifies only one protocol exchange. + +```go +package contracts + +import ( + "fmt" + "strings" + "time" +) + +const SchemaVersion = 1 + +// Kind values match the final topic segment used to route each message. +type Kind string + +const ( + KindTelemetry Kind = "telemetry" + KindCommands Kind = "commands" + KindResponses Kind = "responses" + KindSales Kind = "sales" +) + +// These types define the JSON fields shared by backend and simulator. +type Heartbeat struct { + SchemaVersion int `json:"schema_version"` + MessageID string `json:"message_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + Temperature *float64 `json:"temperature,omitempty"` +} + +type Command struct { + SchemaVersion int `json:"schema_version"` + MessageID string `json:"message_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + CommandID string `json:"command_id"` + Command string `json:"command"` +} + +type CommandResult struct { + SchemaVersion int `json:"schema_version"` + MessageID string `json:"message_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + CommandID string `json:"command_id"` + Status string `json:"status"` + Detail string `json:"detail,omitempty"` +} + +type SalesLine struct { + SKU string `json:"sku"` + Quantity int64 `json:"quantity"` + AmountMinor int64 `json:"amount_minor"` +} + +type SalesReport struct { + SchemaVersion int `json:"schema_version"` + ReportID string `json:"report_id"` + DeviceID string `json:"device_id"` + OccurredAt string `json:"occurred_at"` + Items []SalesLine `json:"items"` +} + +// validID limits identifiers to safe, single-segment topic values. +func validID(value string) bool { + if value == "" || len(value) > 128 { + return false + } + for _, r := range value { + if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._-", r)) { + return false + } + } + return true +} + +// validateEnvelope checks fields shared by every versioned message. +func validateEnvelope(topic Topic, version int, messageID, deviceID, occurredAt string, now time.Time) error { + if version != SchemaVersion { + return fmt.Errorf("unsupported schema_version %d", version) + } + if !validID(messageID) || !validID(deviceID) { + return fmt.Errorf("message_id and device_id must be valid IDs") + } + if topic.DeviceID != deviceID { + return fmt.Errorf("topic device %q does not match payload device_id %q", topic.DeviceID, deviceID) + } + stamp, err := time.Parse(time.RFC3339, occurredAt) + if err != nil { + return fmt.Errorf("occurred_at must be RFC3339: %w", err) + } + if stamp.After(now.Add(30 * time.Second)) { + return fmt.Errorf("occurred_at is more than 30 seconds in the future") + } + return nil +} + +// Each Validate method adds checks specific to one message kind. +func (m Heartbeat) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindTelemetry { + return fmt.Errorf("heartbeat requires telemetry topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if m.Temperature != nil && (*m.Temperature < -50 || *m.Temperature > 100) { + return fmt.Errorf("temperature is outside the supported range") + } + return nil +} + +func (m Command) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindCommands { + return fmt.Errorf("command requires commands topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if !validID(m.CommandID) || m.Command != "ping" { + return fmt.Errorf("command_id is invalid or command is unsupported") + } + return nil +} + +func (m CommandResult) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindResponses { + return fmt.Errorf("command result requires responses topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if !validID(m.CommandID) || (m.Status != "accepted" && m.Status != "completed" && m.Status != "rejected") { + return fmt.Errorf("command result has invalid command_id or status") + } + return nil +} + +func (m SalesReport) Validate(topic Topic, now time.Time) error { + if topic.Kind != KindSales { + return fmt.Errorf("sales report requires sales topic") + } + if err := validateEnvelope(topic, m.SchemaVersion, m.ReportID, m.DeviceID, m.OccurredAt, now); err != nil { + return err + } + if len(m.Items) == 0 || len(m.Items) > 100 { + return fmt.Errorf("sales report must contain 1 to 100 items") + } + for i, item := range m.Items { + if !validID(item.SKU) || item.Quantity <= 0 || item.AmountMinor < 0 { + return fmt.Errorf("item %d has invalid SKU, quantity, or amount_minor", i) + } + } + return nil +} + +// ParseTopic turns broker routing metadata into values handlers can compare. +func ParseTopic(value string) (Topic, error) { + parts := strings.Split(value, "/") + if len(parts) != 5 || parts[0] != "lab" || parts[1] != "v1" || parts[2] != "devices" || !validID(parts[3]) { + return Topic{}, fmt.Errorf("topic must be lab/v1/devices/{deviceID}/{kind}") + } + kind := Kind(parts[4]) + switch kind { + case KindTelemetry, KindCommands, KindResponses, KindSales: + return Topic{DeviceID: parts[3], Kind: kind}, nil + default: + return Topic{}, fmt.Errorf("unsupported topic kind %q", kind) + } +} + +// Topic is the parsed device scope and message kind from an MQTT topic. +type Topic struct { + DeviceID string + Kind Kind +} +``` + +`internal/contracts/decode.go` bounds the payload before decoding, rejects unknown fields, requires the JSON value to consume the entire input, and calls the decoded contract's `Validate` method before returning it. `MQTTMessage` restricts the generic decoder to types with that method. The parsed topic and clock are explicit inputs so validation can check route/body identity and remain deterministic in tests. + +```go +package contracts + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "time" +) + +const MaxPayloadBytes = 16 * 1024 + +// MQTTMessage is implemented by each versioned message contract. +type MQTTMessage interface { + Validate(topic Topic, now time.Time) error +} + +// Decode bounds input, rejects unknown fields, requires one JSON value, +// and validates it against the topic before returning it. +func Decode[T MQTTMessage](payload []byte, topic Topic, now time.Time) (T, error) { + var value T + if len(payload) > MaxPayloadBytes { + return value, fmt.Errorf("payload exceeds %d bytes", MaxPayloadBytes) + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&value); err != nil { + return value, fmt.Errorf("decode JSON: %w", err) + } + var trailing any + // Decode reads one JSON value at a time. A second call distinguishes + // whitespace (EOF) from a second value or malformed trailing bytes. + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return value, fmt.Errorf("payload contains more than one JSON value") + } + return value, fmt.Errorf("trailing data after JSON value: %w", err) + } + if err := value.Validate(topic, now); err != nil { + return value, fmt.Errorf("validate message: %w", err) + } + return value, nil +} +``` + +`internal/contracts/decode_test.go` checks malformed boundaries without a broker: + +```go +package contracts + +import ( + "strings" + "testing" + "time" +) + +// Topic tests reject malformed routes and unsupported message kinds. +func TestParseTopic(t *testing.T) { + topic, err := ParseTopic("lab/v1/devices/device-001/sales") + if err != nil { t.Fatal(err) } + if topic.DeviceID != "device-001" || topic.Kind != KindSales { + t.Fatalf("unexpected topic: %#v", topic) + } + for _, invalid := range []string{"lab/v1/devices/+/sales", "lab/v1/devices/device-001/unknown", "lab/v1/devices/device-001/sales/extra"} { + if _, err := ParseTopic(invalid); err == nil { t.Errorf("ParseTopic(%q) succeeded", invalid) } + } +} + +// These cases cover strict JSON shape and single-value decoding. +func TestDecodeRejectsUnknownAndTrailingValues(t *testing.T) { + topic := Topic{DeviceID: "device-001", Kind: KindTelemetry} + now := time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + if _, err := Decode[Heartbeat]([]byte(`{"schema_version":1,"unknown":true}`), topic, now); err == nil { + t.Fatal("unknown field was accepted") + } + if _, err := Decode[Heartbeat]([]byte(`{"schema_version":1,"message_id":"m-1","device_id":"device-001","occurred_at":"2026-10-13T12:00:00Z"} {"message_id":"m-2"}`), topic, now); err == nil { + t.Fatal("trailing JSON value was accepted") + } +} + +// This case ensures the application limit is applied before decoding. +func TestDecodeRejectsOversizedPayload(t *testing.T) { + payload := []byte(`"` + strings.Repeat("x", MaxPayloadBytes) + `"`) + if _, err := Decode[Heartbeat](payload, topic, now); err == nil { + t.Fatal("oversized payload was accepted") + } +} + +// A body cannot override the device selected by the topic. +func TestCommandRejectsTopicIdentityMismatch(t *testing.T) { + topic, err := ParseTopic("lab/v1/devices/device-001/commands") + if err != nil { t.Fatal(err) } + command := Command{SchemaVersion: SchemaVersion, MessageID: "m-1", DeviceID: "device-002", OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: "c-1", Command: "ping"} + if err := command.Validate(topic, time.Now().UTC()); err == nil { + t.Fatal("topic/body identity mismatch was accepted") + } +} +``` ## 🏷️ 2. Parse topics before routing -> πŸ’‘ **Story question:** A message arrives with both a topic and a payload. Why decide which device and message kind the topic represents before running a handler? +> **❓ Question:** **A message arrives with both a topic and a payload. Why decide which device and message kind the topic represents before running a handler?** -Parsing the topic establishes the intended route from broker metadata; it also gives validation a value to compare against any device identity claimed in the payload. - -Implement a parser accepting exactly `lab/v1/devices/{deviceID}/{kind}`. Initially permit telemetry, commands, responses, and sales. Reject empty device IDs, extra segments, wildcards, and unknown kinds. Return a typed device ID and kind rather than making every handler split strings itself. Extend the lab ACL explicitly for sales only when you start publishing reports. +`contracts.ParseTopic` in `messages.go` accepts exactly `lab/v1/devices/{deviceID}/{kind}`. It rejects empty IDs, extra segments, wildcard characters, and unknown message kinds. Call it before decoding so the validator can compare the route's device ID with the payload claim. Add the sales ACL rules only when you publish sales reports. ## βœ… 3. Decode and validate -> πŸ’‘ **Story question:** JSON decoding succeeded. What else must be true before this message is safe for application state? +> **❓ Question:** **JSON decoding succeeded. What else must be true before this message is safe for application state?** -Structure is only one check. Bound the input, enforce the supported schema and required IDs, validate timestamps and values, and compare topic identity with the body claim. +**πŸ’¬ Answer:** Decoding checks syntax, not whether the message is safe to use. The backend also checks size, schema, required IDs, timestamps, values, and agreement between the topic and claimed device before business handling. -Apply a 16 KiB payload limit before decoding. Use `json.Decoder` with `DisallowUnknownFields`, then require EOF after the first JSON value. Validate schema version 1, required IDs, supported commands, topic/body device agreement, RFC3339 timestamps, and field lengths. Choose and document a future-clock tolerance, such as 30 seconds; inject the current time into validation to keep tests deterministic. Historical sales may legitimately be old, so do not apply heartbeat freshness rules to every report. +`contracts.Decode[T]` in `decode.go` applies the 16 KiB payload limit, rejects unknown JSON fields, requires EOF after one value, and calls `Validate(topic, now)` before returning. `MQTTMessage` restricts `T` to contract types that implement validation. Validation checks schema version 1, required IDs, supported commands, topic/body device agreement, RFC3339 timestamps, and message-specific values. Passing `now` explicitly lets callers use a fixed clock in tests. Historical sales are allowed; the validator does not apply heartbeat freshness rules to reports. -Treat Content Type and Payload Format Indicator as descriptive metadata. Validate bytes even when they claim to contain JSON. A payload device ID is a claim; broker credentials and ACLs supply the separate authorization boundary. +Metadata describes the payload, and the ID in it is only a claim. The backend validates the bytes and compares that ID with the topic. Broker credentials and ACLs separately authorize which sender may publish to the topic. ## πŸ§ͺ 4. Integrate and verify -> πŸ’‘ **Story question:** Where should validation sit so malformed messages cannot trigger a response or change state? +> **❓ Question:** **Where should validation sit so malformed messages cannot trigger a response or change state?** -Run it between transport reception and business handling. Verify both accepted messages and rejection cases, and inspect broker ACLs separately when testing sender permissions. +Run validation after receiving the message but before business handling. Otherwise, a malformed command could already change state or trigger a reply. Test accepted and rejected messages, then test broker ACL permissions separately; application validation does not decide who may publish to a topic. -Replace Section 4's anonymous JSON structs in both programs with these contracts. Validate before updating state or sending a result. Log the error category and bounded identifiers, without dumping rejected payloads. Unsupported or malformed commands must have no simulated effect. +For one sales message, the processing order is: + +```go +topic, err := contracts.ParseTopic(message.Topic) +if err != nil { + return err +} +report, err := contracts.Decode[contracts.SalesReport](message.Payload, topic, time.Now().UTC()) +if err != nil { + return err +} +// Only a validated report reaches the business handler. +``` + +`Decode` reads one JSON value per call. After decoding the expected report, it calls `Decode` again: `io.EOF` means only whitespace remained, another successful decode means there was a second JSON value, and a different error means malformed trailing bytes. Only after that structural check does it call the report's `Validate` method. + +Replace Section 4's anonymous JSON structs in both programs with these contracts. The complete executable files are below; each validates inbound data before logging, changing state, or sending a result. Rejections log a bounded error, not the full payload. + +`cmd/backend/main.go`: + +```go +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "os/signal" + "syscall" + "time" + + "example.com/mqtt5-learning/internal/config" + "example.com/mqtt5-learning/internal/contracts" + "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("load config: %w", err) + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + // Start subscriptions before handling any delivered message. + inbox := make(chan mqtt.Message, 16) + manager, err := mqtt.Start(ctx, cfg, []string{"lab/v1/devices/+/telemetry", "lab/v1/devices/+/responses", "lab/v1/devices/+/sales"}, inbox) + if err != nil { + return fmt.Errorf("start MQTT: %w", err) + } + defer mqtt.Stop(manager) + + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + sequence := 0 + for { + select { + case <-ctx.Done(): + return nil + case message := <-inbox: + if err := handleMessage(message); err != nil { + log.Printf("rejected topic=%q reason=%v", message.Topic, err) + } + case <-ticker.C: + sequence++ + id := fmt.Sprintf("ping-%d-%d", time.Now().UTC().Unix(), sequence) + command := contracts.Command{SchemaVersion: contracts.SchemaVersion, MessageID: id, DeviceID: cfg.MQTT.DeviceID, OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: id, Command: "ping"} + payload, err := json.Marshal(command) + if err != nil { + return fmt.Errorf("encode command: %w", err) + } + if err := mqtt.Publish(ctx, manager, "lab/v1/devices/"+cfg.MQTT.DeviceID+"/commands", payload); err != nil { + log.Printf("command publish failed or uncertain id=%s: %v", id, err) + } + } + } +} + +func handleMessage(message mqtt.Message) error { + // Parse and validate before a message can reach application handling. + topic, err := contracts.ParseTopic(message.Topic) + if err != nil { + return err + } + now := time.Now().UTC() + switch topic.Kind { + case contracts.KindTelemetry: + value, err := contracts.Decode[contracts.Heartbeat](message.Payload, topic, now) + if err != nil { return err } + log.Printf("valid heartbeat device=%s message=%s", value.DeviceID, value.MessageID) + case contracts.KindResponses: + value, err := contracts.Decode[contracts.CommandResult](message.Payload, topic, now) + if err != nil { return err } + log.Printf("command result device=%s command=%s status=%s", value.DeviceID, value.CommandID, value.Status) + case contracts.KindSales: + value, err := contracts.Decode[contracts.SalesReport](message.Payload, topic, now) + if err != nil { return err } + log.Printf("valid sales report device=%s report=%s items=%d", value.DeviceID, value.ReportID, len(value.Items)) + default: + return fmt.Errorf("backend does not consume topic kind %q", topic.Kind) + } + return nil +} +``` + +`cmd/simulator/main.go`: + +```go +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "os/signal" + "syscall" + "time" + + "example.com/mqtt5-learning/internal/config" + "example.com/mqtt5-learning/internal/contracts" + "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) } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + // Publish this device's heartbeat and subscribe only to its command topic. + base := "lab/v1/devices/" + cfg.MQTT.DeviceID + inbox := make(chan mqtt.Message, 16) + manager, err := mqtt.Start(ctx, cfg, []string{base+"/commands"}, inbox) + if err != nil { return fmt.Errorf("start MQTT: %w", err) } + defer mqtt.Stop(manager) + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + sequence := 0 + for { + select { + case <-ctx.Done(): + 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, err := json.Marshal(heartbeat) + if err != nil { return fmt.Errorf("encode heartbeat: %w", err) } + if err := mqtt.Publish(ctx, manager, base+"/telemetry", payload); err != nil { log.Printf("heartbeat publish: %v", err) } + case message := <-inbox: + if message.Retain { log.Printf("reject retained command topic=%q", message.Topic); continue } + if err := respond(ctx, manager, cfg.MQTT.DeviceID, message); err != nil { log.Printf("reject command topic=%q reason=%v", message.Topic, err) } + } + } +} + +func respond(ctx context.Context, manager *autopaho.ConnectionManager, deviceID string, message mqtt.Message) error { + // Check route and body identity before constructing a command result. + topic, err := contracts.ParseTopic(message.Topic) + if err != nil { return err } + command, err := contracts.Decode[contracts.Command](message.Payload, topic, time.Now().UTC()) + if err != nil { return err } + if topic.DeviceID != deviceID { return fmt.Errorf("command is for another device") } + result := contracts.CommandResult{SchemaVersion: contracts.SchemaVersion, MessageID: "result-"+command.CommandID, 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) } + return mqtt.Publish(ctx, manager, "lab/v1/devices/"+deviceID+"/responses", payload) +} +``` ```sh gofmt -w internal/contracts cmd @@ -114,11 +579,11 @@ go test ./internal/contracts go test ./... ``` -`mkdir -p` creates the package directory; you still create the listed `.go` files yourself. `gofmt -w` rewrites source formatting, `go test ./internal/contracts` isolates contract tests, and `go test ./...` checks the whole evolving module for integration regressions. Go discovers test functions named `Test...` in `_test.go` files. No broker is needed for pure parser/validator tests. +`mkdir -p` creates the package directory. `gofmt -w` formats the complete files above. `go test ./internal/contracts` runs the supplied pure parser/validator tests without a broker; `go test ./...` checks the rest of the application too. When adding sales, edit `lab/mosquitto/acl`: add `topic read lab/v1/devices/+/sales` under backend and `topic write lab/v1/devices/device-001/sales` under simulator. Restart the lab broker to read the changed ACL, check logs, and test allowed and cross-device-denied publications. Permissions enable a topic direction; they do not replace sales validation. -Test valid messages, malformed JSON, trailing JSON, unknown fields, missing IDs, unsupported versions, identity mismatch, unknown commands, invalid timestamps, and payloads exactly at and above the limit. Keep the simulator and backend contracts in sync when changing the wire format. +The supplied tests cover topic parsing, malformed JSON boundaries, payload size, and topic/body identity. Add a case when you change a contract so the test captures the new rule. Keep the simulator and backend contracts in sync when changing the wire format. **Done when:** Both programs still exchange ping/pong, every rejection has a predictable reason, and invalid sales or commands leave state unchanged. Explain why schema validation does not authenticate a device. diff --git a/lessons/06-concurrency-cancellation.md b/lessons/06-concurrency-cancellation.md index b01b790..61674b9 100644 --- a/lessons/06-concurrency-cancellation.md +++ b/lessons/06-concurrency-cancellation.md @@ -2,9 +2,9 @@ ## 🎬 Start with the system story -> πŸ’‘ **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? +> **❓ 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. Your application decides how that 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. +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 @@ -20,22 +20,47 @@ flowchart LR 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`. Create files at the project-relative paths shown below, not under `lessons/`. Continue the same application and pinned Paho v0.23.0 dependency. These are guided implementation exercises; attempt each step before requesting a full solution. +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 -For the pinned client's automatic mode, `routePublishPackets` invokes the registered handlers, then calls its session acknowledgement path for nonzero QoS. The boolean and error returned by a handler are passed to later handlers; returning an error does not turn the callback into an application-controlled MQTT NACK or force redelivery. +### πŸ“₯ 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 -receive PUBLISH β†’ copy into RAM queue β†’ callback returns β†’ MQTT ACK - β†’ worker validates/applies effect later +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` changes who releases acknowledgement. For QoS 1, the intended durable-ingestion sequence is receive β†’ validate β†’ commit durable admission β†’ call `received.Client.Ack(received.Packet)`. 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. +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. @@ -56,37 +81,233 @@ Two workers taking 100 ms per item can complete about 20 items/second before oth 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.
-Does returning false or an error prevent the broker from receiving PUBACK? +❓ Question: What does a callback error do to QoS 1 acknowledgement? -Not in the pinned automatic callback path. These return values describe local handler processing. A deliberate acknowledgement policy uses the client's supported manual mechanism and recovery design. It must also handle invalid permanent failures so one bad message cannot occupy capacity forever. +**πŸ’¬ 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.
## ↩️ 1. Trace callback and acknowledgement order -> πŸ’‘ **Story question:** Does returning from the receive callback mean the business operation is durable, or only that the MQTT client may acknowledge delivery? +> **❓ 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. Automatic MQTT acknowledgement can still occur before your application finishes work. This is a deliberate loss window, not reliable ingestion. +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 -> πŸ’‘ **Story question:** When messages arrive faster than handlers can finish, where is excess work stored and how does the application report that capacity is exhausted? +> **❓ 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. -Create `internal/dispatch/pool.go` and `pool_test.go`. Start with two workers and capacity 16. Expose a context-aware submit operation and explicit accepted/full/stopping outcomes. Count accepted, rejected, completed, and failed work. Keep telemetry and important work in separate bounded queues so a heartbeat burst cannot occupy every command slot. +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 -> πŸ’‘ **Story question:** If a worker is still processing when shutdown begins, which components must stay alive until the work finishes or its deadline expires? +> **❓ Question:** **If a worker is still processing when shutdown begins, which components must stay alive until the work finishes or its deadline expires?** -Define buffer ownership, stop admission, drain accepted work, then stop the MQTT producer path in a bounded order. +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. @@ -94,11 +315,11 @@ On shutdown, stop admitting new work and let accepted work finish within a separ ## πŸ”— 4. Integrate and stress it -> πŸ’‘ **Story question:** What observations would show that overload is bounded and shutdown is controlled, rather than merely that the process eventually exits? +> **❓ 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. -Edit the backend and simulator processing loops to submit validated work. Keep transport callbacks short. Add a configurable artificial handler delay and a finite synthetic burst generator; do not create one goroutine per incoming message. Keep these switches in typed TOML + koanf configuration and preserve existing precedence. +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 @@ -114,7 +335,7 @@ Create `internal/dispatch` and `journal` before saving their files; the `mkdir - 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. -Use channels in tests to hold a worker busy deterministically, fill the queue, and assert the next submission fails. Test cancellation, concurrent submissions, pending-work shutdown, and accounting. Avoid tests that rely only on sleeps. +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. diff --git a/lessons/07-command-response.md b/lessons/07-command-response.md index fe7bda4..0e782ae 100644 --- a/lessons/07-command-response.md +++ b/lessons/07-command-response.md @@ -2,7 +2,7 @@ ## 🎬 Start with the system story -> πŸ’‘ **Story question:** The broker acknowledged a backend's command publication. How can the backend match a later device response to the right requestβ€”and what should it report if no response arrives? +> **❓ Question:** **The broker acknowledged a backend's command publication. How can the backend match a later device response to the right requestβ€”and what should it report if no response arrives?** The MQTT acknowledgement covers delivery on one hop. The device's application response is a separate publication. The backend therefore needs application-level command identity, correlation, a pending state registered before publication, and a timeout policy that does not claim the device definitely did nothing. @@ -28,7 +28,7 @@ sequenceDiagram Prerequisite: Sections 5–6 validate messages and bound concurrent processing. 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`. Create files at the project-relative paths shown below, not under `lessons/`. Continue the same application and pinned Paho v0.23.0 dependency. These are guided implementation exercises; attempt each step before requesting a full solution. +Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. Save the complete implementation blocks at the project-relative paths shown below. This lesson provides the command-state and simulator-deduplication code; the exercises ask you to trace its evidence and failure behavior. ## ↩️ MQTT request/response is an application conversation @@ -37,6 +37,7 @@ Response Topic tells the responder where a result should be published. Correlati This fragment is the packet-construction portion of a service method, not a standalone program. `commandJSON` is an encoded validated command and `correlation` is owned by its pending record: ```go +// MQTT 5 carries the reply destination and request identity in packet properties. request := &paho.Publish{ Topic: "lab/v1/devices/device-001/commands", QoS: 1, @@ -52,7 +53,11 @@ request := &paho.Publish{ // Send this packet only after the pending request has been registered. ``` -Pass `request` to the bounded publication operation after registering pending state. The simulator validates `ResponseTopic`, publishes the result to it, and copies Correlation Data unchanged into the result's properties. Correlation bytes are opaque; they need not be text. Do not trust an arbitrary supplied topic as a response destination. +Pass `request` to the bounded publication operation after registering pending state. The simulator reads `ResponseTopic` to find where to publish its result and copies Correlation Data unchanged into the result's properties. Correlation bytes are opaque; they need not be text. + +**❓ Question:** **A command for device-001 asks the simulator to reply on `lab/v1/devices/device-002/responses`. What could go wrong if it publishes there?** + +**πŸ’¬ Answer:** The reply could appear in device-002's message stream. The simulator checks that `ResponseTopic` exactly matches its permitted topic, `lab/v1/devices/device-001/responses`. If it does not match, the simulator rejects the command and publishes no reply. Correlation Data matches a result to a request; it does not authorize the destination. ### ↩️ Trace a fast response and a lost response @@ -77,25 +82,322 @@ Use a synthetic effect counter. Send command C twice with the same content, then Concurrent duplicates need atomic admission: a lookup followed by an unlocked effect allows two workers to both observe "not seen". Reserve the ID with its fingerprint before executing, and define what a second request sees while the first is in progress. A real effect followed by a crash before storing its result remains a separate failure window; the in-memory simulator only demonstrates the idea.
-The broker acknowledged, but no device result arrived. Which status is justified? +❓ Question: The broker acknowledged, but no device result arrived. Which status is justified? -Only the transport acknowledgement is established. The device may be offline, may have rejected the request, may still be executing, or may have executed with a lost response. Once its deadline expires, expose unknown outcome. Retry only under the same business identity and a deliberate idempotency policy. +**πŸ’¬ Answer:** Only the transport acknowledgement is established. The device may be offline, may have rejected the request, may still be executing, or may have executed with a lost response. Once its deadline expires, expose unknown outcome. Retry only under the same business identity and a deliberate idempotency policy.
## πŸ’Ύ 1. Preserve MQTT 5 request metadata -> πŸ’‘ **Story question:** How can a responder know where to send a reply and the requester know which command the reply belongs to? +> **❓ Question:** **How can a responder know where to send a reply and the requester know which command the reply belongs to?** Use the MQTT 5 Response Topic and Correlation Data as transport metadata, while retaining a stable application command ID and checking that the response destination is authorized. -Create `internal/commands/service.go`, `store.go`, and `service_test.go`. Edit `internal/mqtt/client.go` to carry Response Topic and a copied Correlation Data slice in received messages. Add an outbound request form that accepts these properties. Retain the existing bounded publish timeout and reason-code checks. +Save the following complete files as `internal/commands/store.go` and `service.go`. The service stores pending state before publishing, keeps broker acknowledgement separate from device status, and matches a response by device, command ID, and Correlation Data. + +`internal/commands/store.go`: + +```go +package commands + +import ( + "fmt" + "bytes" + "sync" + "time" + + "example.com/mqtt5-learning/internal/contracts" +) + +type Status string + +// Status records application progress independently from BrokerAcked. +const ( + Pending Status = "pending" + Accepted Status = "accepted" + Completed Status = "completed" + Rejected Status = "rejected" + Unknown Status = "unknown" +) + +type Record struct { + CommandID string + DeviceID string + CorrelationData []byte + CreatedAt time.Time + Deadline time.Time + Status Status + BrokerAcked bool + AckReasonCode byte + Detail string +} + +type Store struct { + mu sync.RWMutex + records map[string]Record +} + +// NewStore creates the in-memory command record table. +func NewStore() *Store { return &Store{records: make(map[string]Record)} } + +func key(deviceID, commandID string) string { return deviceID + "/" + commandID } + +func (s *Store) Insert(record Record) error { + s.mu.Lock() + defer s.mu.Unlock() + k := key(record.DeviceID, record.CommandID) + if _, exists := s.records[k]; exists { return fmt.Errorf("command already exists") } + record.CorrelationData = append([]byte(nil), record.CorrelationData...) + s.records[k] = record + return nil +} + +// Get and Update copy correlation bytes so callers cannot mutate stored state. +func (s *Store) Get(deviceID, commandID string) (Record, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + record, ok := s.records[key(deviceID, commandID)] + record.CorrelationData = append([]byte(nil), record.CorrelationData...) + return record, ok +} + +func (s *Store) Update(record Record) error { + s.mu.Lock() + defer s.mu.Unlock() + k := key(record.DeviceID, record.CommandID) + if _, exists := s.records[k]; !exists { return fmt.Errorf("command not found") } + record.CorrelationData = append([]byte(nil), record.CorrelationData...) + s.records[k] = record + return nil +} + +func (s *Store) MarkBrokerAck(deviceID, commandID string, reason byte) (Record, error) { + s.mu.Lock() + defer s.mu.Unlock() + k := key(deviceID, commandID) + record, ok := s.records[k] + if !ok { return Record{}, fmt.Errorf("command not found") } + record.BrokerAcked, record.AckReasonCode = true, reason + s.records[k] = record + return record, nil +} + +// MarkUnknownIfPending preserves any application result that arrived first. +func (s *Store) MarkUnknownIfPending(deviceID, commandID, detail string) (Record, error) { + s.mu.Lock() + defer s.mu.Unlock() + k := key(deviceID, commandID) + record, ok := s.records[k] + if !ok { return Record{}, fmt.Errorf("command not found") } + if record.Status == Pending || record.Status == Accepted { record.Status, record.Detail = Unknown, detail } + s.records[k] = record + return record, nil +} + +// ApplyResult checks both device scope and correlation before changing status. +func (s *Store) ApplyResult(deviceID string, correlation []byte, result contracts.CommandResult) (Record, error) { + s.mu.Lock() + defer s.mu.Unlock() + k := key(deviceID, result.CommandID) + record, ok := s.records[k] + if !ok { return Record{}, fmt.Errorf("no pending command matches result") } + if result.DeviceID != record.DeviceID || !bytes.Equal(correlation, record.CorrelationData) { + return Record{}, fmt.Errorf("result identity or correlation does not match pending command") + } + if record.Status == Completed || record.Status == Rejected { return record, nil } + switch result.Status { + case "accepted": record.Status = Accepted + case "completed": record.Status = Completed + case "rejected": record.Status = Rejected + default: return Record{}, fmt.Errorf("unsupported result status %q", result.Status) + } + record.Detail = result.Detail + s.records[k] = record + return record, nil +} + +func (s *Store) List() []Record { + s.mu.RLock() + defer s.mu.RUnlock() + result := make([]Record, 0, len(s.records)) + for _, record := range s.records { + record.CorrelationData = append([]byte(nil), record.CorrelationData...) + result = append(result, record) + } + return result +} +``` + +`internal/commands/service.go`: + +```go +package commands + +import ( + "context" + "crypto/rand" + "fmt" + "time" + + "example.com/mqtt5-learning/internal/contracts" +) + +type Publisher interface { + PublishCommand(context.Context, string, contracts.Command, string, []byte) (byte, error) +} + +// Service coordinates command identity, storage, publication, and deadlines. +type Service struct { + store *Store + publisher Publisher + now func() time.Time + timeout time.Duration +} + +func NewService(store *Store, publisher Publisher, timeout time.Duration) *Service { + return &Service{store: store, publisher: publisher, now: func() time.Time { return time.Now().UTC() }, timeout: timeout} +} + +func (s *Service) CreateAndPublish(ctx context.Context, deviceID, commandID string) (Record, error) { + // Store pending state first so a fast device response can match immediately. + now := s.now() + correlation := make([]byte, 16) + if _, err := rand.Read(correlation); err != nil { return Record{}, fmt.Errorf("create correlation data: %w", err) } + record := Record{CommandID: commandID, DeviceID: deviceID, CorrelationData: correlation, CreatedAt: now, Deadline: now.Add(s.timeout), Status: Pending} + if err := s.store.Insert(record); err != nil { return Record{}, err } + command := contracts.Command{SchemaVersion: contracts.SchemaVersion, MessageID: "message-"+commandID, DeviceID: deviceID, OccurredAt: now.Format(time.RFC3339), CommandID: commandID, Command: "ping"} + responseTopic := "lab/v1/devices/"+deviceID+"/responses" + ctx, cancel := context.WithDeadline(ctx, record.Deadline) + defer cancel() + reason, err := s.publisher.PublishCommand(ctx, deviceID, command, responseTopic, append([]byte(nil), correlation...)) + if err != nil { + updated, _ := s.store.MarkUnknownIfPending(deviceID, commandID, "publish outcome uncertain: "+err.Error()) + return updated, err + } + updated, err := s.store.MarkBrokerAck(deviceID, commandID, reason) + return updated, err +} + +func (s *Service) HandleResult(deviceID string, correlation []byte, result contracts.CommandResult) (Record, error) { + return s.store.ApplyResult(deviceID, correlation, result) +} + +// Expire records uncertainty; it does not claim that the device did no work. +func (s *Service) Expire(now time.Time) { + for _, record := range s.store.List() { + if (record.Status == Pending || record.Status == Accepted) && !now.Before(record.Deadline) { + _, _ = s.store.MarkUnknownIfPending(record.DeviceID, record.CommandID, "response deadline expired; execution outcome is unknown") + } + } +} + +func (s *Service) Get(commandID string) (Record, bool) { + for _, record := range s.store.List() { + if record.CommandID == commandID { return record, true } + } + return Record{}, false +} + +``` Read `references/paho.golang-v0.23.0/paho/cp_publish.go` for `paho.PublishProperties`. Set `ResponseTopic` to `lab/v1/devices/device-001/responses` and `CorrelationData` to opaque bytes associated with the pending command. Establish the backend response subscription before sending. Register pending state before publishing so an immediate response cannot beat registration. +Save this complete adapter as `internal/mqtt/command_publisher.go`. It checks the response destination before publishing, copies Correlation Data, uses QoS 1, and returns the broker reason code separately from the device result. + +```go +package mqtt + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "example.com/mqtt5-learning/internal/contracts" + "github.com/eclipse/paho.golang/autopaho" + "github.com/eclipse/paho.golang/paho" +) + +type CommandPublisher struct { + Manager *autopaho.ConnectionManager +} + +func (p CommandPublisher) PublishCommand(ctx context.Context, deviceID string, command contracts.Command, responseTopic string, correlation []byte) (byte, error) { + // Authorize the reply destination before putting it in the MQTT packet. + authorized := "lab/v1/devices/" + deviceID + "/responses" + if responseTopic != authorized { return 0, fmt.Errorf("unauthorized response topic") } + if command.DeviceID != deviceID { return 0, fmt.Errorf("command device does not match publish destination") } + payload, err := json.Marshal(command) + if err != nil { return 0, fmt.Errorf("encode command: %w", err) } + operation, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + // Correlation Data is copied so the packet owns stable bytes for this publish. + ack, err := p.Manager.Publish(operation, &paho.Publish{ + Topic: "lab/v1/devices/" + deviceID + "/commands", + QoS: 1, + Payload: payload, + Properties: &paho.PublishProperties{ + ContentType: "application/json", + ResponseTopic: responseTopic, + CorrelationData: append([]byte(nil), correlation...), + }, + }) + if err != nil { return 0, fmt.Errorf("publish command: %w", err) } + if ack == nil { return 0, fmt.Errorf("missing QoS 1 PUBACK") } + if ack.ReasonCode >= 0x80 { return ack.ReasonCode, fmt.Errorf("broker rejected publish: reason 0x%02x", ack.ReasonCode) } + return ack.ReasonCode, nil +} +``` + +The receive callback must copy packet properties while the packet is available. Add these fields to the existing `mqtt.Message` and set them in its receive callback: + +```go +type Message struct { + Topic string + Payload []byte + Retain bool + ResponseTopic string + CorrelationData []byte +} + +// Copy properties alongside payload before the receive packet leaves the callback. +properties := packet.Properties +message := Message{Topic: packet.Topic, Payload: append([]byte(nil), packet.Payload...), Retain: packet.Retain} +if properties != nil { + message.ResponseTopic = properties.ResponseTopic + message.CorrelationData = append([]byte(nil), properties.CorrelationData...) +} +``` + +Save this complete result receiver as `internal/commands/receive.go`. It validates the MQTT topic and payload before applying a response, then matches the received Correlation Data against the pending command. + +```go +package commands + +import ( + "fmt" + "time" + + "example.com/mqtt5-learning/internal/contracts" + "example.com/mqtt5-learning/internal/mqtt" +) + +func HandleMessage(service *Service, message mqtt.Message) error { + // Validate the route and result body before matching the pending request. + topic, err := contracts.ParseTopic(message.Topic) + if err != nil { return err } + if topic.Kind != contracts.KindResponses { return fmt.Errorf("expected a response topic") } + result, err := contracts.Decode[contracts.CommandResult](message.Payload, topic, time.Now().UTC()) + if err != nil { return err } + _, err = service.HandleResult(topic.DeviceID, append([]byte(nil), message.CorrelationData...), result) + return err +} +``` + ## 🧩 2. Model evidence explicitly -> πŸ’‘ **Story question:** What should the backend show after a PUBACK but before any device response arrives? +> **❓ Question:** **What should the backend show after a PUBACK but before any device response arrives?** Represent broker delivery evidence separately from device acceptance, completion, rejection, and timeout/unknown outcome. @@ -105,21 +407,176 @@ Protect concurrent access and define allowed transitions. A completed or rejecte ## ♻️ 3. Make the simulator idempotent -> πŸ’‘ **Story question:** If QoS 1 or an application retry delivers the same command again, how does the device avoid repeating a non-idempotent effect? +> **❓ Question:** **If QoS 1 or an application retry delivers the same command again, how does the device avoid repeating a non-idempotent effect?** Record results by stable command ID and replay the prior result for duplicates instead of applying the simulated effect twice. -Create `internal/simulator/dedup.go` and its tests. Index effects by stable command ID and retain a fingerprint of the command. For an identical duplicate, replay the recorded result without repeating the effect. Reject reuse of the ID with different content. Bound cache lifetime and size; state clearly that restarting the simulator loses this in-memory protection. +Save this complete in-memory simulator cache as `internal/simulator/dedup.go`. It serializes admission and the synthetic effect under one lock, replays an identical result, rejects reuse of an ID with different content, and refuses new work when full rather than evicting live deduplication state. -Allow a Response Topic only when it exactly matches the authorized response topic for this simulator. Correlation is matching data, not authorization. Continue rejecting retained one-shot commands. +```go +package simulator + +import ( + "crypto/sha256" + "encoding/json" + "fmt" + "sync" + + "example.com/mqtt5-learning/internal/contracts" +) + +type cachedResult struct { + fingerprint [32]byte + result contracts.CommandResult +} + +type Deduplicator struct { + mu sync.Mutex + capacity int + results map[string]cachedResult +} + +// NewDeduplicator bounds how many command results this simulator retains. +func NewDeduplicator(capacity int) *Deduplicator { + if capacity < 1 { capacity = 1 } + return &Deduplicator{capacity: capacity, results: make(map[string]cachedResult)} +} + +func AuthorizedResponseTopic(deviceID, requested string) bool { + return requested == "lab/v1/devices/"+deviceID+"/responses" +} + +// Apply executes effect at most once for a device/command ID while this process lives. +func (d *Deduplicator) Apply(command contracts.Command, effect func() (contracts.CommandResult, error)) (contracts.CommandResult, bool, error) { + if effect == nil { return contracts.CommandResult{}, false, fmt.Errorf("effect is required") } + encoded, err := json.Marshal(command) + if err != nil { return contracts.CommandResult{}, false, fmt.Errorf("fingerprint command: %w", err) } + fingerprint := sha256.Sum256(encoded) + key := command.DeviceID + "/" + command.CommandID + d.mu.Lock() + defer d.mu.Unlock() + if previous, ok := d.results[key]; ok { + if previous.fingerprint != fingerprint { return contracts.CommandResult{}, false, fmt.Errorf("command ID was reused with different content") } + return previous.result, true, nil + } + if len(d.results) >= d.capacity { return contracts.CommandResult{}, false, fmt.Errorf("deduplication cache is full") } + result, err := effect() + if err != nil { return contracts.CommandResult{}, false, err } + d.results[key] = cachedResult{fingerprint: fingerprint, result: result} + return result, false, nil +} +``` + +Use the deduplicator in the complete simulator handler below. Save it as `internal/simulator/respond.go`; it checks the requested destination before applying the effect, then publishes the result with the original Correlation Data. + +```go +package simulator + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "example.com/mqtt5-learning/internal/contracts" + "example.com/mqtt5-learning/internal/mqtt" + "github.com/eclipse/paho.golang/autopaho" + "github.com/eclipse/paho.golang/paho" +) + +func Respond(ctx context.Context, manager *autopaho.ConnectionManager, cache *Deduplicator, deviceID string, message mqtt.Message) error { + // Reject retained or misrouted requests before validating or applying an effect. + if message.Retain { return fmt.Errorf("retained commands are 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 } + if !AuthorizedResponseTopic(deviceID, message.ResponseTopic) { return fmt.Errorf("unauthorized response topic") } + // Cache the simulated result by stable command ID to suppress repeat effects. + result, _, err := cache.Apply(command, func() (contracts.CommandResult, error) { + return contracts.CommandResult{SchemaVersion: contracts.SchemaVersion, MessageID: "result-"+command.CommandID, DeviceID: deviceID, OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: command.CommandID, Status: "completed"}, nil + }) + if err != nil { return err } + payload, err := json.Marshal(result) + if err != nil { return fmt.Errorf("encode result: %w", err) } + operation, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + // Reply only to the authorized destination and echo correlation bytes unchanged. + ack, err := manager.Publish(operation, &paho.Publish{Topic: message.ResponseTopic, QoS: 1, Payload: payload, Properties: &paho.PublishProperties{ContentType: "application/json", CorrelationData: append([]byte(nil), message.CorrelationData...)}}) + if err != nil { return fmt.Errorf("publish response: %w", err) } + if ack == nil { return fmt.Errorf("missing response PUBACK") } + if ack.ReasonCode >= 0x80 { return fmt.Errorf("response publish rejected: reason 0x%02x", ack.ReasonCode) } + return nil +} +``` + +Replace the Section 5 simulator entrypoint with this complete version to use the deduplication and response handler. It retains the validated heartbeat publisher and routes every received command through `Respond`. + +```go +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "os/signal" + "syscall" + "time" + + "example.com/mqtt5-learning/internal/config" + "example.com/mqtt5-learning/internal/contracts" + "example.com/mqtt5-learning/internal/mqtt" + "example.com/mqtt5-learning/internal/simulator" +) + +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) } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + base := "lab/v1/devices/" + cfg.MQTT.DeviceID + inbox := make(chan mqtt.Message, 16) + manager, err := mqtt.Start(ctx, cfg, []string{base+"/commands"}, inbox) + if err != nil { return fmt.Errorf("start MQTT: %w", err) } + defer mqtt.Stop(manager) + // Keep recent command results so repeated delivery does not repeat the effect. + cache := simulator.NewDeduplicator(1024) + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + sequence := 0 + for { + select { + case <-ctx.Done(): + 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, err := json.Marshal(heartbeat) + if err != nil { return fmt.Errorf("encode heartbeat: %w", err) } + if err := mqtt.Publish(ctx, manager, base+"/telemetry", payload); err != nil { log.Printf("heartbeat publish: %v", err) } + case message := <-inbox: + if err := simulator.Respond(ctx, manager, cache, cfg.MQTT.DeviceID, message); err != nil { log.Printf("command rejected or response failed topic=%q reason=%v", message.Topic, err) } + } + } +} +``` + +The device checks that the requested response destination exactly matches its own response topic before publishing. Correlation Data matches the result to the pending request; it does not authorize a destination. Retained one-shot commands are rejected before parsing. ## πŸ§ͺ 4. Exercise timing and failures -> πŸ’‘ **Story question:** If a response is late or missing, what can the backend safely conclude, and what should happen to a later result? +> **❓ Question:** **If a response is late or missing, what can the backend safely conclude, and what should happen to a later result?** Exercise timeouts and duplicates while keeping uncertainty visible; a timeout is not proof that the device did nothing. -Replace the backend's unconditional ticker with an explicit finite command exercise or service calls. Send 20 concurrent synthetic pings with separate IDs. Introduce delayed, duplicate, wrong-device, wrong-correlation, and out-of-order responses. Check every command receives only its own evidence. Retry a timed-out request with the same stable ID; never generate a new ID merely because publication timed out. +Use the completed command service and simulator handler to trace 20 concurrent pings with separate IDs. Observe delayed, duplicate, wrong-device, wrong-correlation, and out-of-order responses. Check that each command receives only matching evidence. A retry keeps the same stable command ID; a publish timeout alone does not prove that the device received nothing. ```sh mkdir -p internal/commands internal/simulator @@ -133,7 +590,7 @@ Run `mkdir -p` before saving the service and simulator package files. `gofmt` fo When copying Correlation Data, allocate a new byte slice so later processing owns stable bytes. Comparing correlation bytes means comparing their contents, not pointer addresses. Before updating a shared command map, validate the response and hold its lock only for the state transition. Publish outside the lock so a network delay cannot freeze every command. -Test responses arriving before publish returns, publish failure with uncertain outcome, timeout cleanup, deduplication, and terminal-state protection. Use an injectable clock and fake transport for deterministic service tests. +The implementation handles responses that arrive before the publish call returns because pending state is inserted first and broker acknowledgement updates only its own fields. Inspect the matching, timeout, deduplication, and terminal-state branches in the complete files above; a fake transport can isolate each case without a broker. **Done when:** Concurrent requests match correctly, identical duplicates cause one effect, unexpected destinations are rejected, and late responses follow a documented policy. Explain why PUBACK and Correlation Data cannot prove business completion. diff --git a/lessons/08-http-commands.md b/lessons/08-http-commands.md index af3dc76..4d190b8 100644 --- a/lessons/08-http-commands.md +++ b/lessons/08-http-commands.md @@ -2,7 +2,7 @@ ## 🎬 Start with the system story -> πŸ’‘ **Story question:** A user clicks β€œping device” in an HTTP API. How can the API respond promptly while the MQTT command and device response happen asynchronously? +> **❓ Question:** **A user clicks β€œping device” in an HTTP API. How can the API respond promptly while the MQTT command and device response happen asynchronously?** The HTTP handler validates the request and asks an application service to create a command. The service records its identity and pending status, publishes through the MQTT client, then exposes later device evidence through a status endpoint. HTTP request lifetime, MQTT delivery, and the device operation each have their own acknowledgement and failure boundary. @@ -33,7 +33,7 @@ sequenceDiagram Prerequisite: Section 7 provides a command service and queryable in-memory states. 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`. Create files at the project-relative paths shown below, not under `lessons/`. Continue the same application and pinned Paho v0.23.0 dependency. These are guided implementation exercises; attempt each step before requesting a full solution. +Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. Save the complete HTTP implementation blocks at the project-relative paths shown below. This lesson includes the handlers and server setup; the remaining exercises trace request, retry, and device-result behavior. ## 🌐 Translate evidence between HTTP and MQTT @@ -45,7 +45,9 @@ POST + retry key β†’ admit command C β†’ HTTP 202 + C GET C --------------------------------------------------------------β†’ status C ``` -Do not convert PUBACK directly into HTTP "device completed". Also do not detach work using an unbounded `context.Background()` per command: accepted work needs the service lifecycle, its own deadline, and tracked shutdown. The HTTP request context controls parsing and admission, while the service controls work after acceptance. +**❓ Question:** **The backend publishes a ping and receives PUBACK. Does that mean the device accepted or completed it?** + +**πŸ’¬ Answer:** No. PUBACK confirms the broker acknowledged the backend's publication. It does not confirm device delivery or business acceptance. The HTTP API returns a command ID, and a later status request reports the device's application result if one arrives. The handler uses the request context while parsing. After validation, it passes work to the command service with the application lifecycle context and the command's own deadline. A disconnected caller therefore does not cancel accepted work. Predict a lost HTTP response: the server admits C, but the caller never sees 202 and retries. The same retry key and payload should return C, not create D. Reusing the key with a changed payload is a conflict. Later, scope keys by authenticated caller so two users cannot accidentally share a command record. @@ -58,37 +60,295 @@ No. The lesson's in-memory command and retry-key maps disappear. HTTP 202 descri ## βš™οΈ 1. Add HTTP configuration and handlers -> πŸ’‘ **Story question:** Which process accepts the user's HTTP request, and how does that process reach the command service without putting MQTT logic in the handler? +> **❓ Question:** **Which process accepts the user's HTTP request, and how does that process reach the command service without putting MQTT logic in the handler?** Wire the HTTP server to the application service and preserve bounded parsing, deadlines, and controlled shutdown. -Create `internal/httpapi/server.go`, `handlers.go`, and `handlers_test.go`. Extend `internal/config/config.go` and `configs/lab.toml` with a loopback listener such as `127.0.0.1:18080`, request-body limit, and server timeout settings. Preserve defaults < TOML < environment < explicit CLI flags for supported settings; keep `mqtt.password` TOML-only. Validate these settings before starting listeners. +Save the complete API implementation below as `internal/httpapi/handlers.go` and `server.go`. It binds only to loopback, limits and strictly decodes request bodies, requires an idempotency key, and sets finite HTTP timeouts. The lab listener address is fixed in the complete backend entrypoint below; production authorization is introduced later. -Compose HTTP and MQTT in the existing `cmd/backend/main.go`. Keep the simulator separate. Section 15 introduces separate API and ingestion processes. +`internal/httpapi/handlers.go`: + +```go +package httpapi + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "sync" + + "example.com/mqtt5-learning/internal/commands" +) + +const maxRequestBytes = 4 * 1024 + +// The API depends on command operations, not on the MQTT client itself. +type commandService interface { + CreateAndPublish(ctx context.Context, deviceID, commandID string) (commands.Record, error) + Get(commandID string) (commands.Record, bool) +} + +type retryEntry struct { + fingerprint [32]byte + record commands.Record +} + +type API struct { + commands commandService + serviceCtx context.Context + mu sync.Mutex + retries map[string]retryEntry +} + +// NewAPI keeps retries in memory and ties accepted work to the app lifecycle. +func NewAPI(serviceCtx context.Context, service commandService) *API { + return &API{commands: service, serviceCtx: serviceCtx, retries: make(map[string]retryEntry)} +} + +type createRequest struct { Command string `json:"command"` } +type commandResponse struct { + CommandID string `json:"command_id"` + DeviceID string `json:"device_id"` + Status string `json:"status"` + BrokerAcked bool `json:"broker_acked"` + Detail string `json:"detail,omitempty"` +} + +func (a *API) Handler() http.Handler { + // Keep the public surface small: create commands and read their status. + mux := http.NewServeMux() + mux.HandleFunc("POST /devices/{deviceID}/commands", a.createCommand) + mux.HandleFunc("GET /commands/{commandID}", a.getCommand) + return mux +} + +func (a *API) createCommand(w http.ResponseWriter, r *http.Request) { + // Validate caller input before reserving an idempotency key or publishing. + deviceID := r.PathValue("deviceID") + if !validPathID(deviceID) { http.Error(w, "invalid device ID", http.StatusBadRequest); return } + key := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + if key == "" || len(key) > 128 { http.Error(w, "valid Idempotency-Key required", http.StatusBadRequest); return } + var input createRequest + if err := decodeRequest(w, r, &input); err != nil { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { http.Error(w, "request body too large", http.StatusRequestEntityTooLarge); return } + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if input.Command != "ping" { http.Error(w, "unsupported command", http.StatusBadRequest); return } + // The fingerprint detects reuse of one key for different request content. + fingerprint := sha256.Sum256([]byte(deviceID + "\x00" + input.Command)) + + // This in-memory lesson serializes admissions to make same-key retries atomic. + a.mu.Lock() + defer a.mu.Unlock() + if previous, ok := a.retries[key]; ok { + if previous.fingerprint != fingerprint { http.Error(w, "idempotency key reused with different request", http.StatusConflict); return } + writeCommand(w, previous.record, http.StatusAccepted) + return + } + commandID, err := newCommandID() + if err != nil { http.Error(w, "could not create command ID", http.StatusInternalServerError); return } + record, publishErr := a.commands.CreateAndPublish(a.serviceCtx, deviceID, commandID) + if record.CommandID == "" { + if publishErr != nil { http.Error(w, "command admission unavailable", http.StatusServiceUnavailable); return } + http.Error(w, "command admission failed", http.StatusServiceUnavailable) + return + } + a.retries[key] = retryEntry{fingerprint: fingerprint, record: record} + // The command record exists even if the publish outcome is uncertain. + writeCommand(w, record, http.StatusAccepted) +} + +func (a *API) getCommand(w http.ResponseWriter, r *http.Request) { + record, ok := a.commands.Get(r.PathValue("commandID")) + if !ok { http.Error(w, "command not found", http.StatusNotFound); return } + writeCommand(w, record, http.StatusOK) +} + +// decodeRequest enforces a byte limit, strict fields, and exactly one JSON value. +func decodeRequest(w http.ResponseWriter, r *http.Request, target any) error { + r.Body = http.MaxBytesReader(w, r.Body, maxRequestBytes) + decoder := json.NewDecoder(r.Body) + decoder.DisallowUnknownFields() + if err := decoder.Decode(target); err != nil { return fmt.Errorf("invalid JSON request: %w", err) } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { return errors.New("request must contain one JSON value") } + return fmt.Errorf("trailing request data: %w", err) + } + return nil +} + +func writeCommand(w http.ResponseWriter, record commands.Record, status int) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Location", "/commands/"+record.CommandID) + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(commandResponse{CommandID: record.CommandID, DeviceID: record.DeviceID, Status: string(record.Status), BrokerAcked: record.BrokerAcked, Detail: record.Detail}) +} + +// Path IDs use the same restricted alphabet as contract and topic IDs. +func validPathID(value string) bool { + if value == "" || len(value) > 128 { return false } + for _, r := range value { + if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._-", r)) { return false } + } + return true +} + +func newCommandID() (string, error) { + var raw [16]byte + if _, err := rand.Read(raw[:]); err != nil { return "", err } + return "cmd-" + hex.EncodeToString(raw[:]), nil +} +``` + +`internal/httpapi/server.go`: + +```go +package httpapi + +import ( + "fmt" + "net" + "net/http" + "time" +) + +func NewServer(address string, api *API) (*http.Server, error) { + // The teaching API listens only on loopback and uses finite I/O timeouts. + host, port, err := net.SplitHostPort(address) + if err != nil || (host != "127.0.0.1" && host != "localhost") || port == "" { + return nil, fmt.Errorf("HTTP lab address must be loopback with an explicit port") + } + return &http.Server{ + Addr: address, Handler: api.Handler(), + ReadHeaderTimeout: 5 * time.Second, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + IdleTimeout: 60 * time.Second, + }, nil +} +``` + +The complete final `cmd/backend/main.go` is below. It starts MQTT before HTTP, routes response messages into the command service, and shuts down the HTTP server before disconnecting MQTT. It replaces the Section 5 backend entrypoint. + +```go +package main + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "example.com/mqtt5-learning/internal/commands" + "example.com/mqtt5-learning/internal/config" + "example.com/mqtt5-learning/internal/contracts" + "example.com/mqtt5-learning/internal/httpapi" + "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("load config: %w", err) } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + inbox := make(chan mqtt.Message, 16) + topics := []string{"lab/v1/devices/+/telemetry", "lab/v1/devices/+/responses", "lab/v1/devices/+/sales"} + manager, err := mqtt.Start(ctx, cfg, topics, inbox) + if err != nil { return fmt.Errorf("start MQTT: %w", err) } + defer mqtt.Stop(manager) + + // Wire the command service before exposing its HTTP endpoints. + store := commands.NewStore() + publisher := mqtt.CommandPublisher{Manager: manager} + service := commands.NewService(store, publisher, 15*time.Second) + api := httpapi.NewAPI(ctx, service) + server, err := httpapi.NewServer("127.0.0.1:18080", api) + if err != nil { return err } + serverErrors := make(chan error, 1) + go func() { serverErrors <- server.ListenAndServe() }() + log.Printf("HTTP API listening on %s", server.Addr) + expirationTicker := time.NewTicker(time.Second) + defer expirationTicker.Stop() + + // Keep response processing and expiration active for the server lifetime. + for { + select { + case <-ctx.Done(): + shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := server.Shutdown(shutdown); err != nil { return fmt.Errorf("HTTP shutdown: %w", err) } + return nil + case err := <-serverErrors: + if !errors.Is(err, http.ErrServerClosed) { return fmt.Errorf("HTTP server: %w", err) } + return nil + case <-expirationTicker.C: + service.Expire(time.Now().UTC()) + case message := <-inbox: + topic, err := contracts.ParseTopic(message.Topic) + if err != nil { log.Printf("rejected topic=%q reason=%v", message.Topic, err); continue } + switch topic.Kind { + case contracts.KindResponses: + if err := commands.HandleMessage(service, message); err != nil { log.Printf("rejected response topic=%q reason=%v", message.Topic, err) } + case contracts.KindTelemetry: + var heartbeat contracts.Heartbeat + heartbeat, err = contracts.Decode[contracts.Heartbeat](message.Payload, topic, time.Now().UTC()) + if err != nil { log.Printf("rejected telemetry topic=%q reason=%v", message.Topic, err); continue } + log.Printf("valid heartbeat device=%s message=%s", heartbeat.DeviceID, heartbeat.MessageID) + case contracts.KindSales: + var report contracts.SalesReport + report, err = contracts.Decode[contracts.SalesReport](message.Payload, topic, time.Now().UTC()) + if err != nil { log.Printf("rejected sales topic=%q reason=%v", message.Topic, err); continue } + log.Printf("valid sales report device=%s report=%s", report.DeviceID, report.ReportID) + default: + log.Printf("ignored topic kind=%s", topic.Kind) + } + } + } +} +``` ## πŸ“‹ 2. Define the API contract -> πŸ’‘ **Story question:** What should the caller receive immediately when device work will finish later? +> **❓ Question:** **What should the caller receive immediately when device work will finish later?** Return an accepted command ID, then provide a status resource that reports later evidence. -Implement `POST /devices/{id}/commands` for a JSON command such as `{"command":"ping"}` and `GET /commands/{id}` for status. Limit bodies, reject trailing JSON and unknown fields, validate device IDs, and return JSON errors consistently. Keep the listener local; this lesson does not implement production user authorization. +The complete handlers above implement `POST /devices/{id}/commands` and `GET /commands/{id}`. They limit bodies, reject trailing JSON and unknown fields, validate the path ID, require an idempotency key, and return consistent JSON status. The listener remains local; production user authorization is introduced later. Return HTTP 202 with a command ID and a Location pointing to its status after service admission. Document malformed-input 400, oversized-body 413, unknown-command/device 404 or a consistent chosen alternative, conflict 409, and admission-unavailable 503 behavior. Transport uncertainty after acceptance belongs in command status rather than silently discarding an accepted command. ## 🌐 3. Implement HTTP retry identity -> πŸ’‘ **Story question:** If the server accepted a request but the HTTP response was lost, how can the caller retry without creating a second device action? +> **❓ Question:** **If the server accepted a request but the HTTP response was lost, how can the caller retry without creating a second device action?** Use an HTTP idempotency key with stable request semantics; keep it distinct from the command ID and MQTT packet identifier. -Require an `Idempotency-Key` header for POST. Map the key and request fingerprint to the existing command ID. An identical retry returns the same ID; changed content under the same key returns 409. Serialize concurrent admission for the same key. Bound key retention and document that process restart loses this in-memory guarantee. This key, the stable device command ID, and MQTT packet IDs serve separate purposes. +The handler keeps the key, request fingerprint, and command record together in memory. An identical retry returns the original ID; changed content under that key returns 409. The mutex makes admission atomic in this single-process lesson. A restart loses this map, so this is not a durable idempotency guarantee. The HTTP key, command ID, and MQTT packet ID have different lifetimes and purposes. Use the HTTP context for request parsing and admission. Once accepted, execute under the application's lifecycle and command deadline, so client disconnect does not cancel the command. Inject the command service through a small interface and keep MQTT calls out of handlers. ## πŸ§ͺ 4. Run and verify -> πŸ’‘ **Story question:** What evidence demonstrates each boundary in the round tripβ€”from HTTP acceptance to broker delivery to device result? +> **❓ Question:** **What evidence demonstrates each boundary in the round tripβ€”from HTTP acceptance to broker delivery to device result?** Observe the returned command ID, MQTT acknowledgement, simulator response, and status endpoint separately. @@ -133,7 +393,7 @@ ReadHeaderTimeout bounds header reading; ReadTimeout bounds reading the request `mkdir -p` creates the HTTP package directory, `gofmt` formats edits, and the test/race/vet commands check behavior, executed concurrent access, and suspicious static patterns respectively. Run directory creation before writing the files even though the check block collects the commands together. -Use `httptest` with a fake service for decoding, size limits, idempotency, unavailable admission, unknown IDs, and status responses. Add a service-level check that accepted work continues after the caller context is canceled. +The complete handler performs the same checks an `httptest` would exercise: strict decoding, body limits, idempotency, unavailable admission, unknown IDs, and status responses. Run the supplied local `curl` flow to observe them through the real HTTP server. Accepted work is owned by the command service; device completion arrives later through MQTT. **Done when:** HTTP creates a ping, status reflects device evidence, retries reuse its ID, and closing the caller connection does not erase accepted work. diff --git a/lessons/09-sessions-recovery.md b/lessons/09-sessions-recovery.md index 32fe19c..2102a2a 100644 --- a/lessons/09-sessions-recovery.md +++ b/lessons/09-sessions-recovery.md @@ -6,7 +6,11 @@ Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. ## A connection is temporary; a session may outlive it -The network connection carries packets now. A broker session can preserve subscriptions and eligible pending deliveries after that connection closes. Your Go command map lives in a different place and is not restored by CONNACK. +**❓ Question:** **The simulator loses its network connection after subscribing. What must still exist for a queued ping to reach it when it reconnects?** + +**πŸ’¬ Answer:** The simulator must reconnect with the same Client ID to a broker session that still exists, and that session must contain its QoS 1 subscription. The backend's PUBACK only confirms the broker accepted that publication on this hop; it does not prove the simulator received or completed the ping. Your Go command map lives in a different place and is not restored by CONNACK. + +The network connection carries packets now. A broker session can preserve subscriptions and eligible pending deliveries after that connection closes. | Clean Start on this connection | Prior unexpired session under the same Client ID | Expected Session Present | Meaning | | --- | --- | --- | --- | @@ -20,6 +24,10 @@ In AutoPaho v0.23.0, `CleanStartOnInitialConnection` affects the manager's first ### Work through an offline command +**❓ Question:** **Why does a new Client ID not receive the offline ping sent for the old simulator ID?** + +**πŸ’¬ Answer:** Client ID selects the broker session. A new ID starts or resumes a different session, so it has no old subscription or queued delivery. A `device_id` in the payload is only an application claim and does not select or authenticate a broker session. + At t=0 the simulator has subscribed at QoS 1 with expiry 60. At t=5 it disconnects. At t=10 the backend publishes a nonretained QoS 1 ping. At t=20 the same simulator ID reconnects with Clean Start false. Assuming the broker retained the session, accepted the publication, and no shorter message expiry or resource limit intervened, the subscription allows queued delivery. Reconnect after the session expires and the old queue is gone. Creating a new subscriber after publication is different: it had no prior subscription/session to receive that queued nonretained message. Retained state, studied next, is associated with a topic and can initialize a new subscriber. @@ -44,6 +52,10 @@ Not necessarily. The broker resumed its session while the backend's in-memory ap ## 1. Add session controls +**❓ Question:** **What does `Session Present` tell you after CONNACK, and what does it leave unknown?** + +**πŸ’¬ Answer:** It reports whether the broker resumed prior session state for this connection. It does not report whether the backend's command record survived, whether a device acted, or whether every queued application job completed. + Extend the typed configuration with Clean Start and Session Expiry Interval in seconds, and pass them into AutoPaho. Edit `internal/mqtt/client.go` to log `SessionPresent` from CONNACK without logging credentials. Record connection and subscription readiness separately. Read the pinned `autopaho/auto.go`: `CleanStartOnInitialConnection` applies to the first connection of that manager; automatic reconnects behave differently from constructing a new manager. Use unique stable IDs for backend and simulator. Run one process per client ID. A changing ID creates a different session, while simultaneous clients sharing an ID can take over each other's connection. diff --git a/lessons/10-expiry-presence-state.md b/lessons/10-expiry-presence-state.md index 4b3d592..0818967 100644 --- a/lessons/10-expiry-presence-state.md +++ b/lessons/10-expiry-presence-state.md @@ -4,6 +4,10 @@ Prerequisite: complete the session experiments in Lesson 9. Continue the same la ## 1. Give every timer one responsibility +**❓ Question:** **A ping is queued while a device is offline, then reaches it after the operator has changed the desired setting. Which timer can stop stale delivery, and which check must still run on the device?** + +**πŸ’¬ Answer:** MQTT Message Expiry limits how long the broker may forward the publication. The device must also check the application's absolute deadline before starting the action, because work may have spent time in a local queue or arrived just before expiry. Neither timer can reverse an action already started. + | Timer | What expires | What it cannot prove | | --- | --- | --- | | Go publish context | Your wait for the publication operation | That the device never received it | @@ -13,7 +17,7 @@ Prerequisite: complete the session experiments in Lesson 9. Continue the same la | Keep Alive | Acceptable MQTT connection inactivity | That the application is healthy | | Will Delay | Delay before an eligible Will is published | That the device is physically broken | -A vending configuration command that was useful five minutes ago may be wrong now. Use Message Expiry to limit stale delivery and an application deadline to reject stale execution after delivery. A response timeout merely tells the caller it lacks a result. +A vending configuration command that was useful five minutes ago may be wrong now. Use Message Expiry to limit stale delivery and an application deadline to reject stale execution after delivery. A response timeout merely tells the caller it lacks a result. A PUBACK can arrive before the simulator validates the command; if the application rejects it, that MQTT acknowledgement remains valid for the protocol step it acknowledged. This fragment builds MQTT metadata for a five-second publication lifetime: @@ -31,6 +35,10 @@ Attach `properties` to the outbound PUBLISH. A nil MessageExpiry omits the prope ## 2. Retained state is a latest value, not an event history +**❓ Question:** **A new simulator subscribes and receives a retained `dispense` command from last week. What design choice caused this replay?** + +**πŸ’¬ Answer:** A one-shot action was published with RETAIN set. Retained publications are stored by topic for future subscribers, so use retained messages for replaceable state such as desired configuration and nonretained messages for one-time actions. + A retained publication replaces the stored value for one exact topic. It can initialize a newly created subscription even when that client has never connected before. Offline session queues instead belong to an existing session and its subscriptions. Unsubscribing or deleting a session does not delete a retained value. Use a desired-configuration topic for state such as revision 7 and a temperature setting. Use nonretained command topics for one-shot actions. A retained "dispense" would risk replay when someone subscribes later. @@ -51,6 +59,10 @@ mosquitto_pub -h 127.0.0.1 -p 18883 -V mqttv5 -u simulator -P ' **❓ Question:** **A new device is installed after the broker has started. How can it receive its own credentials and topic permissions without editing files and reloading Mosquitto?** + +> **πŸ’¬ Answer:** A controlled provisioning operator or service authenticates with a separate management identity and asks the broker plugin to create the device client and assign least-privilege roles. The device then connects with its own credentials. The broker checks those credentials and topic permissions; the Go application still validates the delivered topic and payload. + +A provisioning service (or a controlled operator job) authenticates to the broker through a separate management identity. It asks the plugin to create/enable the device client and attach least-privilege roles. The device then connects with its own credentials and can use only the topics those roles permit. + +```mermaid +sequenceDiagram + participant P as Provisioning service + participant B as Mosquitto + Dynamic Security + participant D as Device + participant A as Go application + P->>B: Admin MQTT connection (TLS for remote use) + P->>B: Create device client; assign device role + B-->>P: Management result + D->>B: CONNECT with device credentials + B->>B: Authenticate client and check ACLs + D->>B: PUBLISH only to its permitted device topic + B->>A: Route topic and payload to backend subscription + A->>A: Validate topic, payload, and business rules +``` + +The Go backend does **not** learn the publisher username from an ordinary forwarded `PUBLISH`. The broker enforces the credential-to-topic boundary before routing. The backend still checks that the topic's device ID agrees with the payload and validates the message contract before acting. + +Dynamic Security is a Mosquitto plugin, not a Go MQTT library feature. It manages broker clients, groups, roles, and ACLs through a special MQTT control API. `mosquitto_ctrl` is the command-line client for that API. Changes apply while the broker runs and are saved to the plugin's `dynamic-security.json` state file. [Official Dynamic Security guide](https://mosquitto.org/documentation/dynamic-security/) + +## 🧱 1. Give the plugin its own broker configuration and state + +> **❓ Question:** **What must change before the broker can accept live account-management requests?** + +> **πŸ’¬ Answer:** Mosquitto must load its Dynamic Security plugin and a writable, persistent plugin state file. The static password and ACL files do not automatically become plugin-managed accounts. + +The current lab uses Mosquitto 2.0.22 with `password_file` and `acl_file`; it does **not** load Dynamic Security. Keep the working lab intact while learning. For this exercise create a separate Compose override/configuration and state volume, and mount `dynamic-security.json` at a persistent, broker-writable path. The broker image/version must include the plugin; verify the library path inside the pinned image rather than assuming a distribution path. + +The plugin's broker configuration has this shape (replace the library path with the verified path in your image): + +```conf +listener 1883 +allow_anonymous false +per_listener_settings false +plugin /verified/path/mosquitto_dynamic_security.so +plugin_opt_config_file /mosquitto/data/dynamic-security.json +persistence true +persistence_location /mosquitto/data/ +log_dest stdout +connection_messages true +``` + +For this listener, Dynamic Security supplies both username/password authentication and topic authorization. Do not also configure `password_file` or `acl_file` for the same identities as if they were one unified policy; choose a clear owner for authentication and ACL decisions. The JSON state path must be writable by the broker process and backed by persistent storage with restricted permissions. + +Mosquitto 2.0 requires an initial plugin state file. Generate one with the matching `mosquitto_ctrl` utility, for example: + +```sh +mosquitto_ctrl dynsec init ./lab/dynamic-security.json dynsec-admin +``` + +This creates the initial administrator client interactively. Protect that state file: it contains security configuration and credential hashes. Keep a restricted backup, and test restore as part of broker operations. Do not commit the generated file or put it in an image layer. Newer Mosquitto versions may initialize a missing file differently; follow the documentation for the exact pinned broker version. + +Start only the isolated Dynamic Security lab and verify the plugin loaded in broker logs before creating device accounts. If startup fails, check plugin path, file mount, writable ownership, and version compatibility. Do not fall back to anonymous access to make the experiment pass. + +## πŸͺͺ 2. Model identity and reusable permissions + +> **❓ Question:** **How can a rule be reusable for many devices without giving them access to one another's topics?** + +> **πŸ’¬ Answer:** Give each device a unique authenticated username and use `%u` only as a complete topic segment in role ACL patterns. The broker substitutes the authenticated username when it evaluates the rule. A payload ID or client-selected Client ID is not a substitute for that credential binding. + +Use one unique Dynamic Security client per device username, and bind its configured client ID when the device has a stable ID and the client library can consistently use it. A username is the plugin's primary client lookup key; a client ID alone is not authentication. Never share one fleet credential between devices. + +Create a device role whose ACL patterns substitute `%u` with the authenticated username. In this example the username itself is the device segment, such as `device-001`: + +```text +Role: device + publishClientSend lab/v1/devices/%u/telemetry allow + publishClientSend lab/v1/devices/%u/responses allow + subscribePattern lab/v1/devices/%u/commands allow + publishClientReceive lab/v1/devices/%u/commands allow +``` + +The separation is intentional: `publishClientSend` controls device-to-broker publications; `subscribePattern` controls what filters a device may request; `publishClientReceive` controls what broker-forwarded publications it may receive. A permitted subscription by itself is not always the complete receive policy. + +Create a separate backend role. The backend may publish commands and subscribe to device telemetry/responses, but it should not use the device role. Keep the management administrator separate from both roles. + +Dynamic Security's initial defaults are not deny-all: publishing to the broker and subscribing are denied by default, while receiving broker-forwarded messages is allowed by default. Set `publishClientReceive` to deny explicitly, then grant only the device's command topic through the role. Add only explicit patterns needed by each role; avoid broad grants such as application-wide `#`. Review ACL event types and wildcard behavior in the official guide before relying on a pattern. + +## πŸ› οΈ 3. Create a role and assign a device + +> **❓ Question:** **What exact broker changes happen when inventory says device-001 is ready to join?** + +> **πŸ’¬ Answer:** The operator creates a client keyed by its unique username, binds the intended Client ID when required, and assigns the reusable device role. The device's own credential then selects that client's policy at CONNECT. + +The following illustrates the `mosquitto_ctrl dynsec` command family. `mosquitto_ctrl` connects to the running broker using a separate administrator account; the device credential being created is not the admin credential. + +```sh +# Create a reusable device role once. +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec createRole device + +# Make receive permissions default-deny; the device role grants its command topic below. +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec setDefaultACLAccess publishClientReceive deny + +# Add each role ACL once. Repeat only for ACLs your design actually needs. +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec addRoleACL device publishClientSend 'lab/v1/devices/%u/telemetry' allow 10 +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec addRoleACL device publishClientSend 'lab/v1/devices/%u/responses' allow 10 +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec addRoleACL device subscribePattern 'lab/v1/devices/%u/commands' allow 10 +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec addRoleACL device publishClientReceive 'lab/v1/devices/%u/commands' allow 10 + +# Provision the per-device client, then attach the reusable role. +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec createClient device-001 -i lab-device-001 +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec addClientRole device-001 device 10 +``` + +The tool prompts for the new device password during client creation. Check the exact prompts/options with the pinned `mosquitto_ctrl` version. **Do not put real admin passwords in command arguments**: arguments may appear in process listings or shell history. The `-P "$DYNSEC_ADMIN_PASSWORD"` form above is a disposable-lab placeholder to make the connection explicit, not a production secret-delivery pattern. In a real deployment, run the controller from a restricted job/service and deliver secrets through the platform's protected credential mechanism; use TLS for remote management connections. + +To remove the device's access, disable it immediately; delete it when the account is no longer needed: + +```sh +mosquitto_ctrl -h 127.0.0.1 -p 18883 -u dynsec-admin -P "$DYNSEC_ADMIN_PASSWORD" \ + dynsec disableClient device-001 +``` + +Disabling/deleting a client disconnects its currently connected sessions and prevents future authentication. Record the provisioning request, target device, operator/service identity, result, and time. Make operations idempotent: a retry should detect an existing client/role rather than accidentally broadening permissions. + +## πŸ§ͺ 4. Prove both sides of the boundary + +> **❓ Question:** **How will we know the plugin blocks cross-device access instead of merely starting successfully?** + +> **πŸ’¬ Answer:** Exercise both allowed and denied paths using the actual per-device credentials, then inspect reason codes, delivery, and broker logs. Startup only proves the plugin loaded; it does not prove the topic policy is correct. + +Use separate device and backend credentials and test both permitted and denied operations: + +1. Connect `device-001` with its configured username, password, and client ID. Publish telemetry to `lab/v1/devices/device-001/telemetry`; expect the broker to accept it. +2. Try publishing as that same client to `lab/v1/devices/device-002/telemetry`; expect the broker to deny it. +3. Try subscribing to `lab/v1/devices/device-002/commands`; expect the subscription result to show rejection. +4. Publish an allowed topic with a body claiming a different `device_id`; confirm the Go backend rejects the topic/body mismatch before changing state. +5. Disable the client while it is connected. Confirm the broker disconnects it and a later reconnect is rejected. +6. Restart the broker and verify the client/role/ACL state remains, demonstrating that the plugin state file is persisted. + +Capture broker logs and client-side MQTT reason codes where available. A command's successful exit or a loaded plugin alone is not proof that the ACL is correct. Restore the isolated lab to its prior state when finished. + +## πŸ—οΈ Production design checklist + +- Keep device credentials unique, revocable, rotated, and delivered through a protected provisioning channel. +- Keep Dynamic Security administrator credentials out of device firmware and ordinary message-processing code. A dedicated provisioning service/job with narrowly controlled access is usually easier to audit than giving the backend's main runtime account administrator power. +- Expose the broker management listener only where needed; require TLS and restrict network reachability. On a local-only lab, loopback access is appropriate. +- Grant per-role least privilege, separate publish from subscribe/receive, and review every wildcard. Keep default-deny behavior. +- Use a stable identity mapping. `%u` is useful only when the authenticated username is unique and provisioned to represent that device; never trust an arbitrary client-selected topic or JSON ID as identity proof. +- Persist and back up Dynamic Security state securely. Test recovery, broker upgrades, plugin compatibility, revocation, and credential rotation. +- Audit provisioning changes and alert on repeated denied access. Protect the audit stream from modification by device clients. +- Consider an external authentication/authorization service when policy must integrate with a device registry or central identity system. Dynamic Security is a broker-local role system, not a device inventory database. + +**Done when:** You can explain who authenticates a device, which component decides its topic permissions, what the backend can infer from the delivered topic, and how revocation affects an already-connected client. + +**Official references:** [Dynamic Security plugin](https://mosquitto.org/documentation/dynamic-security/), [Mosquitto authentication methods](https://mosquitto.org/documentation/authentication-methods/), [Dynamic Security `mosquitto_ctrl` commands](https://www.mosquitto.org/man/mosquitto_ctrl_dynsec-1.html). diff --git a/lessons/16-mqtt5-controls.md b/lessons/16-mqtt5-controls.md index c2cc7a9..95bc9b0 100644 --- a/lessons/16-mqtt5-controls.md +++ b/lessons/16-mqtt5-controls.md @@ -4,6 +4,10 @@ Study this after Sections 12–15 in [COURSE.md](../COURSE.md), or read it earli ## 1. Subscription options change what arrives +**❓ Question:** **A device publishes a retained status update and then subscribes to that same topic. Which option controls retained replay, and which option controls the RETAIN bit on forwarded live traffic?** + +**πŸ’¬ Answer:** Retain Handling controls retained replay at subscription time. Retain As Published controls whether the broker preserves the incoming RETAIN bit when forwarding a live publication. Neither option changes the broker's ACL decision or filters payload identity. + These fields belong to each `paho.SubscribeOptions` entry, not to the PUBLISH payload: | Field | Meaning | Experiment | @@ -20,6 +24,10 @@ An UNSUBSCRIBE removes a broker subscription after its UNSUBACK exchange; it doe ## 2. Shared subscriptions distribute work +**❓ Question:** **Two ingestion processes share a subscription group, and a response reaches the process that did not create the pending waiter. What fails?** + +**πŸ’¬ Answer:** The receiving process may have no matching in-memory command record. A shared subscription distributes each matching delivery to a group member; it does not share process memory or guarantee device affinity. Keep response correlation state shared or route responses to a common durable store. + Ordinary subscriptions on two clients each receive a matching publication. With `$share/ingest/lab/v1/devices/+/telemetry`, both clients join the `ingest` group for that filter and the broker selects a group member for a matching delivery. Give each process a unique Client ID; sharing a group is different from sharing a Client ID. Create two authorized lab consumers, check their SUBACK results, then publish numbered synthetic telemetry. Record which instance handles each ID. Stop one consumer and repeat. Use a fresh lab group/session for the experiment so old offline state does not obscure the result. Verify the broker's shared-subscription support and effective ACL behavior before drawing conclusions. @@ -28,6 +36,10 @@ Distribution is not guaranteed round-robin, per-device affinity, or business exa ## 3. Distinguish wire limits from application limits +**❓ Question:** **The backend sets Receive Maximum to four. Does that cap the number of telemetry jobs waiting in its Go worker queue at four?** + +**πŸ’¬ Answer:** No. This CONNECT limit constrains the broker's concurrent QoS 1/2 publications awaiting protocol acknowledgement on the broker-to-client direction. If callbacks acknowledge quickly after queueing, the application queue can continue to grow until its own bound is reached. + The CONNECT properties describe what this client can receive. The CONNACK properties describe what the server can receive or supports. Direction matters: changing the client's Receive Maximum constrains broker-to-client QoS 1/2 inflight publications, not the backend's outgoing command rate. Add this field to your existing `autopaho.ClientConfig`; it is a fragment, with `net/url` and Paho imports supplied by that file: @@ -55,6 +67,10 @@ For packet-size testing, use harmless payloads below and above a deliberately sm ## 4. Properties carry context, not trust +**❓ Question:** **A simulator sets User Property `producer=device-001`. Has the broker authenticated that device identity?** + +**πŸ’¬ Answer:** No. User Properties are forwarded metadata. The broker authenticates the MQTT client and applies its topic ACL; the application validates the topic, payload, and any metadata it chooses to use. Never treat a property or payload ID as proof of the publisher. + User Properties are UTF-8 key/value pairs and may repeat a key. Use them for synthetic trace context or schema hints, with size/count limits. They are not authenticated claims by themselves and are not HTTP headers that the broker interprets for your application. ```go @@ -72,6 +88,10 @@ Subscription Identifiers let a subscriber label a subscription and identify whic ## 5. Topic aliases optimize one connection +**❓ Question:** **The broker restarted and the MQTT session resumed. Can the client safely publish using an alias-only packet from before the restart?** + +**πŸ’¬ Answer:** No. Topic Alias mappings belong to one network connection and direction. Reestablish the mapping after reconnect; session resumption does not restore it. + An alias associates a small integer with a topic name for one direction of one connection. The sender first establishes the mapping with the topic and alias; subsequent publications may use the alias to avoid resending the long topic string. This reduces bytes, not permission checks or payload size. The receiver advertises its Topic Alias Maximum. Zero or absent support must not be treated as permission to send arbitrary aliases. A reconnect loses the mapping even when an MQTT session resumes. Never send an alias-only first publication after reconnect. @@ -80,6 +100,10 @@ Read the pinned `paho/extensions/topicaliases/` implementation and its tests, pl ## 6. Read failures as protocol evidence +**❓ Question:** **A QoS 1 PUBACK has reason code `0x10` (No matching subscribers). Did the device accept the command?** + +**πŸ’¬ Answer:** No. The reason describes the broker's handling of that publication hop. It is not an application response from the device. The command service must wait for and validate a correlated device response, and must preserve an unknown outcome if none arrives. + Reason codes belong to a packet context. A SUBACK contains a result for each filter. A QoS 1 PUBACK describes the immediate publication hop. A DISCONNECT may describe why the broker ended a connection. None is a device business status. | Observation | Interpretation | Next action | diff --git a/lessons/README.md b/lessons/README.md index 8c7b932..47ab90e 100644 --- a/lessons/README.md +++ b/lessons/README.md @@ -17,8 +17,9 @@ Follow these lessons in order from the learning project root. Read the concept a 9. [Learn MQTT 5 sessions through failure experiments](09-sessions-recovery.md) 10. [Expiry, presence, and retained state](10-expiry-presence-state.md) 11. [MQTT security boundaries](11-security-boundaries.md) +11B. [Provision device ACLs with Mosquitto Dynamic Security](11b-dynamic-security.md) β€” optional implementation extension to Section 11 -Continue Sections 12–15 in the course before implementing [Section 16 β€” MQTT 5 controls](16-mqtt5-controls.md). The course's application/persistence/operations Sections 12–15 and 17–18 remain guided checklists, not complete reference programs. +Continue with course Sections 12–15 after the security lessons, then implement [Section 16 β€” MQTT 5 controls](16-mqtt5-controls.md). Course sections 12–15 and 17–18 remain guided checklists, not complete reference programs. Paho is pinned to **v0.23.0**, the selected course baseline. Use the matching checkout under `references/paho.golang-v0.23.0/`. This release requires Go 1.24 or newer; the lessons' Go 1.25 baseline is compatible. @@ -51,6 +52,7 @@ Checked local Markdown links, source paths, code fences, and expandable answer b | Clean Start, Session Expiry, Session Present, recovery stores | 9 | Predict network versus process restart outcomes | | Message Expiry, retained deletion, Will/Will Delay, freshness | 10 | Observe stale command suppression and presence limitations | | TLS, authentication, ACLs, identity and response destinations | 1 and 11 | Fail each security boundary deliberately | +| Live per-device provisioning and revocation | 11B (optional) | Deny cross-device topics, then disable a connected client | | Subscription options, unsubscribe, shared consumers | 16 | Compare live/replayed messages and consumer distribution | | Receive Maximum, packet size, properties, aliases, capabilities | 16 | Record packet direction, negotiated limit, and measured behavior | | Docker mount/volume identity, recreation, overrides, crashes | Broker operations | Explain exactly which state survives each command |