mqtt-5-learning/teaching/course-source/07-command-response.md

🔁 Section 7 — Build reliable command and response behavior

🎬 Start with the system 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?

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.

sequenceDiagram
    participant H as HTTP/backend caller
    participant G as Go command service
    participant B as Broker
    participant D as Device
    H->>G: Request command
    G->>G: Create command ID and pending waiter
    G->>B: PUBLISH command with correlation data
    B-->>G: PUBACK: broker hop accepted
    B->>D: PUBLISH command
    D->>D: Validate, deduplicate, act
    D->>B: PUBLISH correlated application result
    B->>G: PUBLISH result
    G->>G: Match identity/correlation; update state
    G-->>H: Accepted / completed / rejected / unknown

Before each implementation step, ask: Which event changes command state: broker acknowledgement, device acceptance, device completion, or timeout? Keep those events separate. Follow Lesson 0 for the full client and broker lifecycle.

Prerequisite: Section 5 validates messages, Section 5A turns a heartbeat into an application status view, and Section 6 bounds 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.

Scope: This section uses an in-memory command store. The record is registered before publication to prevent a fast-response race, but process restart loses pending and completed command records. An HTTP request is not introduced until Section 8 and is not durable acceptance until Sections 13–14 add database state and recoverable publication intent.

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

Response Topic tells the responder where a result should be published. Correlation Data lets the requester recognize which pending request it belongs to. The broker forwards these properties; it does not run the command, create a result, or enforce your timeout.

This first command is intentionally a harmless ping: it teaches how to track request identity, broker acknowledgement, device result, timeout, and duplicate delivery. It does not change a vending machine's catalog or configuration. Sections 10 and 12 build desired state and versioned vending updates on top of this command lifecycle.

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:

commandTopic, err := contracts.DeviceTopic("device-001", contracts.KindCommands)
if err != nil { return err }
responseTopic, err := contracts.DeviceTopic("device-001", contracts.KindResponses)
if err != nil { return err }

// MQTT 5 carries the reply destination and request identity in packet properties.
request := &paho.Publish{
    Topic:   commandTopic.String(),
    QoS:     1,
    Retain:  false,
    Payload: commandJSON,
    Properties: &paho.PublishProperties{
        ContentType:     "application/json",
        ResponseTopic:   responseTopic.String(),
        CorrelationData: append([]byte(nil), correlation...),
    },
}

// Send this packet only after the pending request has been registered.

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

1. Backend establishes its response subscription and checks SUBACK.
2. Backend stores the expected device, command ID, correlation, and deadline.
3. Backend publishes. The simulator can respond before the publish call returns.
4. Backend validates and applies device evidence independently of the broker acknowledgement.
5. If the waiting deadline expires, the backend records unknown outcome. In this reference store, a later validated result with matching device and correlation can resolve unknown; completed and rejected remain terminal.

Registering after step 3 loses the race with a fast response. Holding the command-map mutex across publishing can block the callback needed to record that response. A context timeout stops a local wait; it cannot recall bytes already delivered to a device.

Identifier	Owner and lifetime	Why it cannot replace the others
MQTT packet ID	Library, per protocol exchange/session direction, reusable	Identifies a transport handshake, not a permanent command
Correlation Data	Request/response mechanism	Matches a response; does not authenticate it or prevent repeated effects
Command ID	Application, stable across retries	Keys the dedup record and business status

🧭 Define the application state model

The intended lifecycle is created → pending delivery → accepted → completed/rejected; a response deadline moves unresolved work to unknown. A valid late device result may resolve unknown in this reference implementation, while completed and rejected remain terminal. BrokerAcked is an independent transport fact and does not advance that lifecycle.

The example inserts a record directly as Pending before calling PUBLISH, so the moment of creation is not a separately queryable status. This keeps the in-memory teaching implementation small while preserving the important transition and failure boundaries. The exercises should state this modeling choice and its limits.

♻️ Make duplicate behavior observable

Use a synthetic effect counter. Send command C twice with the same content, then C with changed content. Expect one effect, a replayed recorded result, then a conflict. Also send the same C in two separate publish calls: application duplicates need not carry MQTT DUP. The DUP flag describes protocol retransmission, so deduplicating only packets with DUP set is wrong.

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.

❓ Question: The broker acknowledged, but no device result arrived. Which status is justified?

💬 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

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

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.

🔍 store.go — file overview

This file stores in-memory command records and enforces correlation and status transitions. Its record keeps broker acknowledgement separately from device-reported status.

internal/commands/store.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
}

🏗️ NewStore — initialize command state

Summary: Creates the map that indexes records by device and command ID.

Details: The store is process-local; a restart loses pending and completed records. Section 13 replaces this volatile lifetime for durable command state.

➕ Insert, Get, and Update — manage owned records

Summary: Add a unique command, retrieve a copy, or replace an existing record.

Details: The mutex protects the map, and each method copies CorrelationData so a caller cannot mutate stored bytes after the call. The composite key prevents two devices using the same command ID from colliding.

📬 MarkBrokerAck — record transport evidence

Summary: Stores that the broker acknowledged the command publication and its reason code.

Details: This method changes BrokerAcked, not the device Status. A successful broker acknowledgement does not mean the device accepted or completed the command.

⏳ MarkUnknownIfPending — preserve uncertain outcomes

Summary: Marks a pending or accepted command as unknown while preserving terminal results.

Details: A response may arrive before a publish timeout or expiry path runs. The status check prevents a later timeout from overwriting Completed or Rejected.

🔗 ApplyResult — match and apply device evidence

Summary: Requires a known command, matching device, and matching Correlation Data before changing its application status.

Details: Correlation bytes are compared by content. Terminal results do not regress when a duplicate accepted response arrives. Correlation is a matching mechanism; topic ACLs and device identity validation remain separate authorization checks.

📋 List — copy a snapshot for iteration

Summary: Returns all command records without exposing stored correlation byte slices.

Details: The snapshot lets expiry logic iterate without holding the store lock while it performs subsequent updates.

internal/commands/service.go:

🔍 service.go — file overview

This file coordinates command state and the transport adapter. It creates and stores pending state before publishing, handles device results, and records deadline expiry as uncertainty.

package commands

import (
	"context"
	"crypto/rand"
	"fmt"
	"time"

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

type Publisher interface {
	PublishCommand(context.Context, string, contracts.Command, contracts.TopicName, []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) {
	responseTopic, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	if err != nil { return Record{}, fmt.Errorf("build response topic: %w", err) }
	// 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"}
	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
}

🧰 NewService — connect command dependencies

Summary: Binds a command store, publisher, clock, and response timeout into one service.

Details: The service coordinates business state; the Publisher interface hides the MQTT implementation so state transitions can be reasoned about independently of the client.

🚀 CreateAndPublish — register before sending

Summary: Creates correlation bytes and a pending record, stores it, then publishes with a deadline.

Details: Inserting first handles a fast response that arrives before PublishCommand returns. If publication errors or times out, the broker may still have received it, so the method marks the outcome unknown rather than asserting the device did nothing. A successful return records the broker acknowledgement separately.

📩 HandleResult — apply a matched result

Summary: Delegates response identity and correlation checks to the store.

Details: Call it only after the receiver has parsed the topic and validated the result contract. The store then verifies that this application result belongs to the expected pending command.

⏰ Expire — mark response timeouts as unknown

Summary: Scans commands whose deadlines passed and changes pending/accepted records to unknown.

Details: Expiry describes what the backend knows, not what the device did. A late device result can still arrive and needs the course's documented late-result policy.

🔎 Get — query a command record

Summary: Finds a command by its application command ID.

Details: The method returns the application record for HTTP/status use; it does not query the device or the broker.

Read references/paho.golang-v0.23.0/paho/cp_publish.go for paho.PublishProperties. Build the response destination with contracts.DeviceTopic(deviceID, contracts.KindResponses), then pass its string form to Paho's ResponseTopic field. Attach CorrelationData as 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.

🔍 command_publisher.go — file overview

This file adapts the command service to AutoPaho. It checks the requested reply destination, encodes the command, and publishes one MQTT 5 request with Response Topic and Correlation Data.

Save this complete adapter as internal/mqtt/command_publisher.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 contracts.TopicName, correlation []byte) (byte, error) {
	// Authorize the reply destination before putting it in the MQTT packet.
	authorized, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	if err != nil { return 0, fmt.Errorf("build authorized response topic: %w", err) }
	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") }
	commandTopic, err := contracts.DeviceTopic(deviceID, contracts.KindCommands)
	if err != nil { return 0, fmt.Errorf("build command topic: %w", err) }
	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: commandTopic.String(),
		QoS: 1,
		Payload: payload,
		Properties: &paho.PublishProperties{
			ContentType: "application/json",
			ResponseTopic: responseTopic.String(),
			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
}

📤 PublishCommand — send the transport request

Summary: Verifies device and reply-topic agreement, encodes the command, and waits for a bounded QoS 1 publish result.

Details: It copies correlation bytes into the packet so the request owns stable data. A returned reason code is broker-hop evidence only. It is returned separately from the device's eventual application result, which arrives later on the response subscription.

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:

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

🔍 receive.go — file overview

This file is the boundary between an incoming response packet and command state. It checks topic kind and payload contract before passing device identity and Correlation Data to the service.

Save this complete result receiver as internal/commands/receive.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
}

📩 HandleMessage — validate and match a result

Summary: Parses a response topic, decodes a CommandResult, then asks the service to match and apply it.

Details: Malformed or wrong-kind messages stop before the command store is touched. The device ID comes from the parsed topic and is checked against the body and pending record. Correlation Data is copied before it enters state handling.

🧩 2. Model evidence explicitly

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

Store command ID, expected device, correlation bytes, creation/deadline times, business status, and independent transport acknowledgement evidence. Start with created → pending delivery → accepted → completed or rejected. A ping simulator can send accepted followed by completed. Response timeout means unknown outcome; it does not establish that no effect happened.

Protect concurrent access and define allowed transitions. A completed or rejected command must not regress on a duplicate accepted response. Compare command identity, topic/body device identity, and correlation before applying a response. Keep timed-out records for a bounded retention period even after removing their waiters, so status can explain the uncertainty.

♻️ 3. Make the simulator idempotent

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

🔍 dedup.go — file overview

This file records a bounded set of command IDs and their outcomes in simulator memory. It ensures the demonstration does not repeat the same synthetic effect while this process remains alive.

Save this complete in-memory simulator cache as internal/simulator/dedup.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 {
	expected, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	return err == nil && requested == expected.String()
}

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

🏗️ NewDeduplicator — create bounded replay state

Summary: Allocates the cache and raises a too-small capacity to one.

Details: Capacity prevents this example from growing its map forever. The cache is volatile; restarting the simulator forgets the IDs, so it does not provide durable idempotency.

🔐 AuthorizedResponseTopic — constrain reply routing

Summary: Allows only the response topic assigned to this device ID.

Details: Correlation bytes match a response to a request, but do not authorize where the response is sent. The simulator checks the destination before publishing.

♻️ Apply — execute or replay one command

Summary: Fingerprints the command, checks the device/command key under a lock, and runs the effect only when no prior result exists.

Details: An identical retry receives the stored result; reusing an ID for different content fails; a full cache refuses new work instead of evicting a result that still protects against a duplicate. Holding the lock across the synthetic effect makes admission atomic in this small example. A real network call or long-running operation should not run under this lock.

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.

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
}

📩 Respond — validate, deduplicate, and reply

Summary: Rejects retained or misrouted commands, validates the contract and reply destination, obtains a cached result, then publishes it with the request's correlation bytes.

Details: Route and destination checks happen before the simulated effect so an unauthorized request cannot trigger it. cache.Apply ensures a repeated command reuses its first result while this process remains alive. The bounded publish context prevents a stuck response operation from occupying a worker indefinitely. PUBACK for the result still proves only the broker hop; the backend must receive and match the response separately.

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.

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()
	commandFilter, err := contracts.DeviceSubscription(cfg.MQTT.DeviceID, contracts.KindCommands)
	if err != nil { return err }
	telemetryTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindTelemetry)
	if err != nil { return err }
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(ctx, cfg, []contracts.TopicFilter{commandFilter}, 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, telemetryTopic, 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) }
		}
	}
}

🧭 main and run — operate the simulator lifecycle

Summary: main reports a fatal error; run starts the device client, creates the bounded deduplication cache, publishes heartbeats, and delegates commands to Respond.

Details: One cache is created per process so duplicate deliveries share replay state. The ticker is stopped and MQTT manager closed on return. The receive loop invokes Respond outside the callback, keeping packet reception separate from validation, effect handling, and reply publication. Restarting the simulator clears this cache, which is why it demonstrates process-local idempotency rather than durable device guarantees.

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

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

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.

mkdir -p internal/commands internal/simulator

gofmt -w internal cmd

go test -race ./...

Run mkdir -p before saving the service and simulator package files. gofmt formats your edits and go test -race ./... executes the module's tests with concurrency instrumentation. A fake transport is a small in-process implementation of the caller's transport interface: it lets tests deliberately return errors or responses without needing a broker. An injected clock makes deadline tests deterministic without waiting in real time.

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.

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.

✅ Result — the story so far

The backend now treats a ping as a tracked command: it records the command ID, expected device, correlation data, and deadline before publishing. The simulator checks the command and authorized response destination, applies the simulated effect once for that stable ID, and publishes a result. The backend matches that result to the pending command and updates its status. A broker PUBACK still means only that the broker acknowledged its hop; a missing response leaves the outcome unknown, not necessarily unexecuted.

Previous: 06-concurrency-cancellation.md. Next: 08-http-commands.md.