mqtt-5-learning/teaching/lessons/0010-expiry-presence-state.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Lesson 10 — Expiry, Presence, and Desired State</title>
  <link rel="stylesheet" href="../assets/lesson.css"><script src="../assets/quiz.js" defer></script>
</head>
<body><main class="page">
  <p class="eyebrow">Go MQTT Vending Backend · Lesson 10 · 12 minutes</p>
  <h1>Give stale work an expiry; treat presence as evidence</h1>
  <p class="dek">Use separate clocks for publication lifetime, application deadlines, sessions, and freshness. Keep desired state distinct from one-shot commands.</p>
  <div class="meta"><span class="pill">🎯 One skill: assign timer ownership</span><span class="pill">🧠 Predict stale delivery</span><span class="pill">🧪 Simulator + broker</span></div>
  <section class="card callout"><p class="question">❓ Start with the system story</p><p>A vending machine has been offline for an hour. A queued “apply price update” arrives after it reconnects, but the price is obsolete. Which timer prevents stale delivery, and which check prevents late execution after delivery?</p><p>Answer before reading: broker policy and device policy are separate checks.</p></section>
  <h2>⏱️ One timer, one responsibility</h2>
  <div class="two-col"><article class="card"><h3>MQTT Message Expiry</h3><p>Bounds how long a PUBLISH remains eligible for onward delivery. The broker may discard expired queued work.</p></article><article class="card"><h3>Application deadline</h3><p>An absolute deadline in the command contract lets the device reject work that is already delivered but too old to execute.</p></article><article class="card"><h3>Session Expiry</h3><p>Controls how long the broker may preserve the client's session after disconnect.</p></article><article class="card"><h3>Heartbeat freshness</h3><p>Backend receive time measures how recently an application report arrived. It is not a broker connection probe or health diagnosis.</p></article></div>
  <p>Keep Alive detects a lack of MQTT control traffic; it does not show that the application is producing heartbeats. A publish context timeout ends your wait, not necessarily the remote operation. Will Delay delays an eligible Will; it does not certify why the device disappeared.</p>
  <form class="card quiz" data-quiz data-correct="b" data-explanation="Message Expiry bounds queued delivery; the device deadline rejects work that is already delivered but too old to execute." data-retry="One policy bounds broker delivery; the other is enforced by the application before the action.">
    <p><strong>🧠 Check the boundary:</strong> What combination protects against a command that waits in the broker and then spends time in a local queue?</p>
    <label><input type="radio" name="q" value="a"> Session Expiry and Retain Handling protect delayed command execution</label><br><label><input type="radio" name="q" value="b"> MQTT Message Expiry and device deadline protect delayed command execution</label><br><label><input type="radio" name="q" value="c"> Keep Alive and Will Delay protect delayed command execution</label><br><button type="submit">Check answer</button><p data-feedback aria-live="polite"></p>
  </form>
  <h2>📌 Retained desired state is not an event queue</h2>
  <p>A retained publication stores the latest value for one topic and can initialize a new subscriber. It is useful for replaceable desired configuration, such as catalog revision 9. A retained “dispense” or “run calibration once” action can replay to a later subscriber, so one-shot commands should be nonretained.</p>
  <p>For desired state, keep the highest revision at the simulator and reject revision 8 after revision 9. The device reports the revision it applied. A PUBACK only confirms the relevant QoS hop; it does not prove application. An empty retained publication removes the stored value for that topic; it does not retract copies already delivered.</p>
  <h3>📄 Complete file: <code>internal/contracts/topic.go</code></h3>
  <p><strong>File overview:</strong> This is the single source of truth for concrete publish names, controlled subscription filters, and inbound topic parsing. The added kinds support presence and desired/reported state.</p>
  <p><strong><code>DeviceTopic</code> / <code>DeviceFilter</code> summary:</strong> Build validated exact names and fleet filters. <strong><code>ParseTopic</code> summary:</strong> Derive the device and kind from the shared prefix and reject unknown routes.</p>
  <details class="card" open><summary><strong>🧩 Full replacement source</strong></summary><pre><code>package contracts

import (
	"fmt"
	"strings"
)

type TopicName string
type TopicFilter string
type Kind string

const (
	KindTelemetry Kind = "telemetry"
	KindCommands  Kind = "commands"
	KindResponses Kind = "responses"
	KindSales     Kind = "sales"
	KindPresence  Kind = "presence"
	KindDesired   Kind = "desired"
	KindReported  Kind = "reported"
)

const topicPrefix = "lab/v1/devices/"

type Topic struct {
	DeviceID string
	Kind     Kind
}

func (t TopicName) String() string   { return string(t) }
func (f TopicFilter) String() string { return string(f) }

func DeviceTopic(deviceID string, kind Kind) (TopicName, error) {
	if !validID(deviceID) {
		return "", fmt.Errorf("invalid device ID %q", deviceID)
	}
	if !supportedKind(kind) {
		return "", fmt.Errorf("unsupported topic kind %q", kind)
	}
	return TopicName(topicPrefix + deviceID + "/" + string(kind)), nil
}

func DeviceFilter(kind Kind) (TopicFilter, error) {
	if !supportedKind(kind) {
		return "", fmt.Errorf("unsupported topic kind %q", kind)
	}
	return TopicFilter(topicPrefix + "+/" + string(kind)), nil
}

func DeviceSubscription(deviceID string, kind Kind) (TopicFilter, error) {
	topic, err := DeviceTopic(deviceID, kind)
	if err != nil {
		return "", err
	}
	return TopicFilter(topic.String()), nil
}

func ParseTopic(value string) (Topic, error) {
	if !strings.HasPrefix(value, topicPrefix) {
		return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix)
	}
	parts := strings.Split(strings.TrimPrefix(value, topicPrefix), "/")
	if len(parts) != 2 || !validID(parts[0]) {
		return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix)
	}
	kind := Kind(parts[1])
	if !supportedKind(kind) {
		return Topic{}, fmt.Errorf("unsupported topic kind %q", kind)
	}
	return Topic{DeviceID: parts[0], Kind: kind}, nil
}

func supportedKind(kind Kind) bool {
	switch kind {
	case KindTelemetry, KindCommands, KindResponses, KindSales,
		KindPresence, KindDesired, KindReported:
		return true
	default:
		return false
	}
}

func validID(value string) bool {
	if value == "" || len(value) &gt; 128 {
		return false
	}
	for _, r := range value {
		if !(r &gt;= 'a' &amp;&amp; r &lt;= 'z' || r &gt;= 'A' &amp;&amp; r &lt;= 'Z' ||
			r &gt;= '0' &amp;&amp; r &lt;= '9' || strings.ContainsRune("._-", r)) {
			return false
		}
	}
	return true
}
</code></pre></details>
  <h2>🧩 Complete file: expiry and command deadline policy</h2>
  <p><strong>File overview:</strong> Create <code>internal/mqtt/expiry.go</code> to build a QoS 1 publication with explicit expiry, and <code>internal/deviceview/deadline.go</code> to reject stale commands at execution time. Each file is complete and has its own imports, types, and errors.</p>
  <details class="card" open><summary><strong>📄 <code>internal/mqtt/expiry.go</code> — complete file</strong></summary><pre><code>package mqtt

import (
	"fmt"

	"example.com/mqtt5-learning/internal/contracts"
	"github.com/eclipse/paho.golang/paho"
)

// ExpiringPublish builds a publication whose broker-queued lifetime is bounded.
// The device must still check its application deadline before executing it.
func ExpiringPublish(topic contracts.TopicName, payload []byte, expirySeconds uint32) (*paho.Publish, error) {
	if topic == "" {
		return nil, fmt.Errorf("topic is required")
	}
	if expirySeconds == 0 {
		return nil, fmt.Errorf("expirySeconds must be greater than zero")
	}
	seconds := expirySeconds
	return &amp;paho.Publish{
		Topic: topic.String(), Payload: append([]byte(nil), payload...), QoS: 1,
		Properties: &amp;paho.PublishProperties{
			ContentType: "application/json",
			MessageExpiry: &amp;seconds,
		},
	}, nil
}
</code></pre></details>
  <details class="card" open><summary><strong>📄 <code>internal/deviceview/deadline.go</code> — complete file</strong></summary><pre><code>package deviceview

import (
	"fmt"
	"time"
)

// CheckCommandDeadline rejects a command before the simulator starts its effect.
// The absolute deadline protects against time spent in local queues after delivery.
func CheckCommandDeadline(deadline string, now time.Time) error {
	if deadline == "" {
		return fmt.Errorf("command deadline is required")
	}
	when, err := time.Parse(time.RFC3339, deadline)
	if err != nil {
		return fmt.Errorf("parse command deadline: %w", err)
	}
	if !now.Before(when) {
		return fmt.Errorf("command deadline has passed")
	}
	return nil
}
</code></pre></details>
  <h2>🧩 Complete file: simulator's monotonic desired state</h2>
  <p><strong>File overview:</strong> Create <code>internal/deviceview/revisions.go</code> to keep a per-device revision and prevent older desired state from replacing newer applied state.</p>
  <p><strong><code>Apply</code> summary:</strong> Lock, compare the incoming revision, and update only when it is newer. Return an explicit outcome so duplicate and stale deliveries are observable.</p>
  <details class="card" open><summary><strong>📄 <code>internal/deviceview/revisions.go</code> — complete file</strong></summary><pre><code>package deviceview

import "sync"

type RevisionOutcome uint8

const (
	RevisionApplied RevisionOutcome = iota + 1
	RevisionDuplicate
	RevisionStale
)

type RevisionStore struct {
	mu      sync.Mutex
	current map[string]uint64
}

func NewRevisionStore() *RevisionStore {
	return &amp;RevisionStore{current: make(map[string]uint64)}
}

func (s *RevisionStore) Apply(deviceID string, revision uint64) RevisionOutcome {
	s.mu.Lock()
	defer s.mu.Unlock()
	current, exists := s.current[deviceID]
	if exists &amp;&amp; revision == current {
		return RevisionDuplicate
	}
	if exists &amp;&amp; revision &lt; current {
		return RevisionStale
	}
	s.current[deviceID] = revision
	return RevisionApplied
}

func (s *RevisionStore) Current(deviceID string) (uint64, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	revision, ok := s.current[deviceID]
	return revision, ok
}
</code></pre></details>
  <h2>🧭 File overview: <code>internal/mqtt/presence.go</code></h2>
  <p>Create this complete helper file for the Last Will portion. <code>OfflineWill</code> builds the message prepared during CONNECT; <code>OnlinePresence</code> builds the retained publication sent after a successful connection. Both use one validated presence topic and stable device/boot identity.</p>
  <h3>📴 <code>OfflineWill</code> — short summary</h3><p>Returns a retained QoS 1 offline Will with a five-second Will Delay property.</p>
  <h3>🟢 <code>OnlinePresence</code> — short summary</h3><p>Creates the retained online publication for the same device and boot identity.</p>
  <details class="card" open><summary><strong>📄 <code>internal/mqtt/presence.go</code> — complete file</strong></summary><pre><code>package mqtt

import (
	"encoding/json"
	"fmt"
	"time"

	"example.com/mqtt5-learning/internal/contracts"
	"github.com/eclipse/paho.golang/paho"
)

type Presence struct {
	SchemaVersion int    `json:"schema_version"`
	MessageID     string `json:"message_id"`
	DeviceID      string `json:"device_id"`
	BootID        string `json:"boot_id"`
	Status        string `json:"status"`
	OccurredAt    string `json:"occurred_at"`
}

func OfflineWill(deviceID, bootID string) (*paho.WillMessage, *paho.WillProperties, error) {
	if deviceID == "" || bootID == "" {
		return nil, nil, fmt.Errorf("device ID and boot ID are required")
	}
	topic, err := contracts.DeviceTopic(deviceID, contracts.KindPresence)
	if err != nil {
		return nil, nil, fmt.Errorf("build presence topic: %w", err)
	}
	body, err := json.Marshal(Presence{
		SchemaVersion: contracts.SchemaVersionOne,
		MessageID: "will-" + bootID, DeviceID: deviceID, BootID: bootID,
		Status: "offline", OccurredAt: time.Now().UTC().Format(time.RFC3339),
	})
	if err != nil { return nil, nil, fmt.Errorf("encode offline Will: %w", err) }
	delay := uint32(5)
	return &amp;paho.WillMessage{Topic: topic.String(), QoS: 1, Retain: true, Payload: body},
		&amp;paho.WillProperties{WillDelayInterval: &amp;delay, ContentType: "application/json"}, nil
}

func OnlinePresence(deviceID, bootID string, now time.Time) (*paho.Publish, error) {
	if deviceID == "" || bootID == "" {
		return nil, fmt.Errorf("device ID and boot ID are required")
	}
	topic, err := contracts.DeviceTopic(deviceID, contracts.KindPresence)
	if err != nil { return nil, fmt.Errorf("build presence topic: %w", err) }
	body, err := json.Marshal(Presence{
		SchemaVersion: contracts.SchemaVersionOne,
		MessageID: "online-" + bootID, DeviceID: deviceID, BootID: bootID,
		Status: "online", OccurredAt: now.UTC().Format(time.RFC3339),
	})
	if err != nil { return nil, fmt.Errorf("encode online presence: %w", err) }
	return &amp;paho.Publish{
		Topic: topic.String(), Payload: body, QoS: 1, Retain: true,
		Properties: &amp;paho.PublishProperties{ContentType: "application/json"},
	}, nil
}
</code></pre></details>
  <div class="card"><h3>🔎 Details: a Will timestamp is not a failure timestamp</h3><p>The Will body is prepared at CONNECT, so <code>OccurredAt</code> above is the preparation time. The backend should record its own receipt time and use it when evaluating freshness. Include a boot ID so stale presence from a previous process can be distinguished. The simulator's planned shutdown must publish retained offline presence before sending normal DISCONNECT if the application needs that final state; graceful DISCONNECT suppresses the Will.</p></div>
  <h2>🧪 Predict, then observe</h2>
  <ol><li>Set a persistent simulator session, disconnect it, publish a QoS 1 command with a short Message Expiry Interval, then reconnect after expiry.</li><li>Repeat with a longer expiry but an already-passed application deadline. The simulator should receive and reject it as stale.</li><li>Publish desired revision 9, then revision 8; verify reported state remains at 9.</li><li>Compare graceful shutdown, abrupt simulator termination, reconnect before Will Delay, and broker stop. Record the exact fault and evidence.</li><li>Clear the retained value and subscribe fresh. Verify no retained replay.</li></ol>
  <div class="card"><h3>❤️ Presence has limits</h3><p>Retained “online” is the broker's latest stored assertion. A Last Will is published by the broker after an eligible unexpected loss; it is not sent by the dead process. A clean DISCONNECT normally suppresses the Will. Combine connection events with backend-received heartbeat freshness and device-reported health rather than treating one as the other.</p><p><strong>What could go wrong?</strong> A stale retained online value can outlive a graceful shutdown; Will Delay may interact with session expiry; a command delivered just before its expiry may still sit in a local queue; clocks can disagree. Record backend receive time and define the device deadline policy.</p></div>
  <h2>🔎 Details: ownership and limitations</h2><p><code>ExpiringPublish</code> creates protocol metadata only; wire it into the existing publisher and handle the returned PUBACK as transport evidence. <code>CheckCommandDeadline</code> must run immediately before the simulator effect, after local queue wait. <code>RevisionStore</code> is intentionally RAM-only, so a simulator restart loses its revision; PostgreSQL or device flash is needed for durable state. The will is a broker publication prepared during CONNECT; normal DISCONNECT suppresses it. These helpers do not replace ACL updates, topic/body validation, or device response reporting.</p>
  <h2>✅ Done when</h2><p>Demonstrate expired queued work, device-side stale rejection, retained replacement/deletion, stale revision rejection, and the difference between graceful and unexpected disconnect evidence.</p>
  <details class="card"><summary><strong>📋 Short self-check answers</strong></summary><ol><li>Broker expiry limits onward delivery; an application deadline limits whether the device starts the action.</li><li>Retained messages represent latest state, not historical event delivery.</li><li>Heartbeat freshness says a recent report arrived, not that the device is healthy or the broker currently has a connection.</li></ol></details>
  <p>📚 Detailed experiments: <a href="../course-source/10-expiry-presence-state.md">Section 10 source lesson</a>. Primary references: <a href="https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html">OASIS MQTT 5.0 Standard</a>, <a href="https://pkg.go.dev/github.com/eclipse/paho.golang/paho@v0.23.0">Paho v0.23.0 API</a>, and <a href="https://mosquitto.org/man/mosquitto_pub-1.html">Mosquitto publish client manual</a>.</p>
  <p>Ask me to trace an expiry or presence result if the packet evidence differs from your prediction.</p>
  <footer><p>Previous: <a href="0009-session-recovery.html">Lesson 9 — session recovery</a> · Next: <a href="0011-security-boundaries.html">Lesson 11 — security boundaries</a>.</p></footer>
</main></body></html>