mqtt-5-learning/teaching/lessons/0009-session-recovery.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Lesson 9 — MQTT Sessions and Recovery</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 9 · 10 minutes</p>
  <h1>Recover broker session work across disconnects</h1>
  <p class="dek">Make MQTT session policy visible in the connection code, then test what survives network loss and process restart.</p>
  <div class="meta"><span class="pill">🎯 One skill: trace session ownership</span><span class="pill">🧠 Predict reconnect behavior</span><span class="pill">🧪 Disposable broker lab</span></div>
  <section class="card callout"><p class="question">❓ Start with the system story</p><p>A simulator disconnects after subscribing. The backend publishes a QoS 1 ping. What must the broker still have when the simulator reconnects?</p><p>Predict the Client ID, session expiry, subscription, and CONNACK Session Present before reading the implementation.</p></section>
  <h2>🗺️ File overview</h2>
  <p><code>internal/mqtt/client.go</code> owns the AutoPaho connection, MQTT subscriptions, and the inbox callback. This lesson edits that file to use a stable-session profile and record the broker's CONNACK evidence. The backend and simulator already load distinct, stable Client IDs from configuration.</p>
  <h3>🔌 <code>Start</code> — short summary</h3>
  <p>Creates AutoPaho with Clean Start disabled for the initial connection and a 60-second Session Expiry Interval. It logs Session Present on every connection and restores the configured subscriptions.</p>
  <h3>📤 <code>Publish</code> and <code>Stop</code> — short summary</h3>
  <p><code>Publish</code> sends QoS 1 and checks the PUBACK reason. <code>Stop</code> performs a bounded graceful disconnect.</p>
  <h2>🧩 Complete file: <code>internal/mqtt/client.go</code></h2>
  <p>Replace the file with this full version. It preserves the lesson's bounded inbox and adds session logging; it does not make application command records durable.</p>
  <pre><code>package mqtt

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"time"

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

type Message struct {
	Topic   string
	Payload []byte
	Retain  bool
}

const sessionExpirySeconds uint32 = 60

func Start(ctx context.Context, cfg config.Config, topics []contracts.TopicFilter, inbox chan&lt;- Message) (*autopaho.ConnectionManager, error) {
	broker, err := url.Parse(cfg.MQTT.URL)
	if err != nil {
		return nil, fmt.Errorf("parse broker URL: %w", err)
	}
	initialSubscription := make(chan error, 1)
	manager, err := autopaho.NewConnection(ctx, autopaho.ClientConfig{
		ServerUrls:                    []*url.URL{broker},
		KeepAlive:                     20,
		CleanStartOnInitialConnection: false, // Resume this Client ID's session when one exists.
		SessionExpiryInterval:         sessionExpirySeconds,
		ConnectUsername:               cfg.MQTT.Username,
		ConnectPassword:               []byte(cfg.MQTT.Password),
		ClientID:                      cfg.MQTT.ClientID,
		OnConnectionUp: func(cm *autopaho.ConnectionManager, ack *paho.Connack) {
			if ack != nil {
				log.Printf("mqtt connected client_id=%s session_present=%t", cfg.MQTT.ClientID, ack.SessionPresent)
			}
			subCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
			defer cancel()
			subscriptions := make([]paho.SubscribeOptions, 0, len(topics))
			for _, topic := range topics {
				subscriptions = append(subscriptions, paho.SubscribeOptions{
					Topic: topic.String(), QoS: 1, RetainAsPublished: true,
				})
			}
			suback, subErr := cm.Subscribe(subCtx, &amp;paho.Subscribe{Subscriptions: subscriptions})
			if subErr == nil {
				if suback == nil || len(suback.Reasons) != len(topics) {
					subErr = fmt.Errorf("missing or incomplete SUBACK")
				} else {
					for i, reason := range suback.Reasons {
						if reason &gt; 1 {
							subErr = fmt.Errorf("subscription %s rejected: 0x%02x", topics[i], reason)
							break
						}
					}
				}
			}
			if subErr != nil {
				log.Printf("mqtt subscription failed: %v", subErr)
			} else {
				log.Printf("mqtt subscriptions ready client_id=%s count=%d", cfg.MQTT.ClientID, len(topics))
			}
			select {
			case initialSubscription &lt;- subErr:
			default: // Reconnect callbacks must never block on this one-time startup signal.
			}
		},
		OnConnectionDown: func() bool {
			log.Printf("mqtt connection down client_id=%s; reconnecting", cfg.MQTT.ClientID)
			return true
		},
		OnConnectError: func(err error) { log.Printf("mqtt connect error: %v", err) },
		OnClientError: func(err error) { log.Printf("mqtt client error: %v", err) },
		OnServerDisconnect: func(packet *paho.Disconnect) {
			log.Printf("mqtt server disconnect reason=0x%02x", packet.ReasonCode)
		},
		OnPublishReceived: []func(paho.PublishReceived) (bool, error){
			func(received paho.PublishReceived) (bool, error) {
				packet := received.Packet
				message := Message{
					Topic: packet.Topic, Payload: append([]byte(nil), packet.Payload...), Retain: packet.Retain,
				}
				select {
				case inbox &lt;- message:
				default:
					// Automatic acknowledgement may still complete this delivery.
					log.Printf("mqtt inbox full; application message dropped topic=%s", packet.Topic)
				}
				return true, nil
			},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("start AutoPaho: %w", err)
	}
	startup, cancel := context.WithTimeout(ctx, cfg.ConnectTimeout())
	defer cancel()
	if err := manager.AwaitConnection(startup); err != nil {
		Stop(manager)
		return nil, fmt.Errorf("await MQTT connection: %w", err)
	}
	select {
	case err := &lt;-initialSubscription:
		if err != nil {
			Stop(manager)
			return nil, fmt.Errorf("await MQTT subscriptions: %w", err)
		}
	case &lt;-startup.Done():
		Stop(manager)
		return nil, fmt.Errorf("await MQTT subscriptions: %w", startup.Err())
	}
	return manager, nil
}

func Publish(ctx context.Context, manager *autopaho.ConnectionManager, topic contracts.TopicName, payload []byte) error {
	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	ack, err := manager.Publish(operation, &amp;paho.Publish{
		Topic: topic.String(), Payload: payload, QoS: 1,
		Properties: &amp;paho.PublishProperties{ContentType: "application/json"},
	})
	if err != nil {
		return fmt.Errorf("publish: %w", err)
	}
	if ack == nil {
		return fmt.Errorf("missing QoS 1 publish acknowledgement")
	}
	if ack.ReasonCode &gt;= 0x80 {
		return fmt.Errorf("publish rejected: reason=0x%02x", ack.ReasonCode)
	}
	log.Printf("mqtt publish acknowledged topic=%s reason=0x%02x", topic, ack.ReasonCode)
	return nil
}

func Stop(manager *autopaho.ConnectionManager) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := manager.Disconnect(ctx); err != nil {
		log.Printf("MQTT shutdown did not finish within deadline: %v", err)
	}
}
</code></pre>
  <div class="card"><h3>🔎 Details: the session guarantees</h3><ul><li><code>CleanStartOnInitialConnection: false</code> asks AutoPaho to resume this Client ID's session when available. AutoPaho uses Clean Start false on subsequent reconnects regardless; process restart creates a new manager and applies the configured initial behavior again.</li><li><code>SessionExpiryInterval</code> is seconds. A 60-second session can disappear after the interval or broker loss if persistence is disabled.</li><li><code>SessionPresent</code> reports broker session resumption only. This callback logs it separately from subscription readiness.</li><li>Subscribing again on connection-up is deliberate for a simple lab, but duplicate SUBSCRIBE behavior must be observed. It does not register another local Go callback.</li><li>The helper still uses automatic acknowledgement and a RAM inbox. Session continuity cannot repair a message dropped after callback admission, and it cannot restore the backend's in-memory command record.</li></ul></div>
  <h2>🧪 Run the experiment matrix</h2>
  <p>Rebuild both programs after changing the constant or Clean Start setting. Use one simulator per Client ID. Record prediction, broker settings, CONNACK Session Present, SUBACK results, and command delivery in <code>journal/09-sessions.md</code>.</p>
  <ol><li>Connect and subscribe, stop the simulator gracefully, publish a QoS 1 ping while it is offline, then reconnect with the same Client ID before 60 seconds.</li><li>Repeat with Clean Start true on the initial connection, then with a changed Client ID, then after the expiry interval.</li><li>Compare broker restart with persistence disabled and enabled, using the isolated Compose override described in the full source lesson.</li><li>Keep the backend command ID and simulator effect counter separate from Session Present.</li></ol>
  <h2>🧭 Session queue versus retained latest value</h2>
  <table><thead><tr><th>Delivery mechanism</th><th>What it represents</th><th>Who can receive it later</th></tr></thead><tbody><tr><td>Queued nonretained QoS 1 publication</td><td>One event for an existing matching session subscription</td><td>That session before expiry, if broker state and QoS permit</td></tr><tr><td>Retained publication</td><td>The latest value stored for a topic</td><td>A new subscriber, even with a fresh Client ID</td></tr></tbody></table>
  <div class="card"><p><strong>Safe lab comparison:</strong> After the queued-ping run, reconnect the original simulator and keep it online so the retained probe cannot also queue in its offline session. Publish a synthetic retained ping to its lab command topic with the authorized lab backend identity; the Lesson 7 simulator must reject it before any effect. Subscribe a fresh observer Client ID with the lab simulator identity; it should receive the stored value immediately without a prior session. Clear it immediately with an empty retained publication (`mosquitto_pub -r -n`), then verify a new subscription gets no replay. Use no physical device or production credential, and do not leave retained commands in the lab.</p><p>A retained message is the latest value, not event history. Do not use retained publications for one-shot commands.</p></div>
  <div class="card"><h3>✅ Done when</h3><p>You can explain which subscriptions and eligible queued messages the broker preserved and which Go state was lost. Session Present is never used as proof of command completion.</p></div>
  <details class="card"><summary><strong>📋 Short self-check answers</strong></summary><ol><li>Stable Client ID, resumable unexpired session, matching QoS subscription, eligible nonexpired QoS publication, and available broker state.</li><li>Session Present means the broker resumed session state, not application memory or device effect.</li><li>A fresh session after process restart is different from an automatic reconnect within the existing manager.</li></ol></details>
  <p>📚 Full session experiment details: <a href="../course-source/09-sessions-recovery.md">Section 9 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> and <a href="https://pkg.go.dev/github.com/eclipse/paho.golang/autopaho@v0.23.0">Paho AutoPaho v0.23.0 API</a>.</p>
  <p>If the observed session behavior differs from your prediction, ask me to trace the CONNECT/CONNACK and SUBSCRIBE/SUBACK evidence with you.</p>
  <footer><p>Previous: <a href="0008-http-acceptance-vs-completion.html">Lesson 8</a> · Next: <a href="0010-expiry-presence-state.html">Lesson 10 — expiry and presence</a>.</p></footer>
</main></body></html>