mqtt-5-learning/teaching/lessons/0013-persist-reports-command-state.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Lesson 13 — Persist Reports and Command 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 13 · 15 minutes</p>
  <h1>Commit business state before acknowledging important work</h1>
  <p class="dek">Use PostgreSQL uniqueness and transactions to make report effects durable across duplicate delivery and backend restart.</p>
  <div class="meta"><span class="pill">🎯 One skill: choose the durable acceptance boundary</span><span class="pill">🧠 Predict crash recovery</span><span class="pill">🧪 Dedicated lab database only</span></div>
  <section class="card callout"><p class="question">❓ Start with the system story</p><p>The backend commits a QoS 1 sales report to PostgreSQL, then crashes before sending PUBACK. What should happen when the broker delivers the report again?</p><p>Pause before continuing: identify the durable idempotency key and the order of transaction and acknowledgement.</p></section>
  <h2>📥 Acknowledge after the policy boundary</h2>
  <div class="flow" aria-label="durable ingestion sequence"><div class="step">Receive QoS 1<br><strong>validate</strong></div><span class="arrow">→</span><div class="step">PostgreSQL<br><strong>dedupe + effect transaction</strong></div><span class="arrow">→</span><div class="step">COMMIT<br><strong>durable outcome</strong></div><span class="arrow">→</span><div class="step">Manual Ack<br><strong>check error</strong></div></div>
  <p>With Paho v0.23.0 automatic acknowledgement, a callback error does not request MQTT retry; Paho acknowledges QoS 1 after callback return. Enabling manual acknowledgement changes who calls Ack, but returning an error still does not cause retry by itself. Keep the packet/client acknowledgement handle until the durable policy decision, not just a copied payload in a RAM queue.</p>
  <form class="card quiz" data-quiz data-correct="b" data-explanation="If commit succeeded but PUBACK was lost, redelivery is safe: the unique business key finds the existing outcome, no effect repeats, and the backend can acknowledge the duplicate." data-retry="Consider the crash after COMMIT but before Ack, then ask what survives a process restart.">
    <p><strong>🧠 Check crash recovery:</strong> PostgreSQL committed a report, but the backend crashed before PUBACK. What should replay do?</p>
    <label><input type="radio" name="q" value="a"> Apply the sale again because MQTT redelivered it</label><br><label><input type="radio" name="q" value="b"> Find the stable unique key, return the stored outcome without repeating the effect, then acknowledge</label><br><label><input type="radio" name="q" value="c"> Treat the earlier callback's error as an automatic retry guarantee</label><br><button type="submit">Check answer</button><p data-feedback aria-live="polite"></p>
  </form>
  <h2>🔁 Let the database arbitrate duplicates</h2>
  <p>Use a unique business key such as <code>(device_id, report_id)</code>. In one transaction, insert the report identity, apply its business effect, and store the application outcome needed for duplicate replay. The unique constraint handles two concurrent workers and remains after restart; an in-memory map cannot.</p>
  <p>If the same key arrives with materially different content, record a conflict instead of silently giving one ID two meanings. MQTT packet IDs are not business IDs. QoS 2 cannot make a PostgreSQL transaction and a physical vending action exactly once.</p>
  <h2>🧭 File overview: <code>internal/persistence/sales.go</code></h2>
  <p>Create this complete repository file for durable sales ingestion. The caller opens a PostgreSQL <code>*sql.DB</code> using the project's pgx stdlib driver and validates the report with <code>contracts.Decode</code> first. This file owns the transaction boundary: report identity and sale lines commit together.</p>
  <h3>🏗️ <code>Migrate</code> — short summary</h3><p>Creates a report table with a unique business key and a child line table.</p>
  <h3>💾 <code>SaveReport</code> — short summary</h3><p>Inserts the report identity, detects duplicate or conflicting content, inserts lines for a new report, and commits once.</p>
  <details class="card" open><summary><strong>🧩 Complete file contents</strong></summary><pre><code>package persistence

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

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

type ReportOutcome uint8

const (
	ReportSaved ReportOutcome = iota + 1
	ReportDuplicate
)

var ErrReportIDConflict = errors.New("report ID was reused with different content")

type SalesRepository struct {
	db *sql.DB
}

func NewSalesRepository(db *sql.DB) (*SalesRepository, error) {
	if db == nil {
		return nil, fmt.Errorf("database connection is required")
	}
	return &amp;SalesRepository{db: db}, nil
}

func (r *SalesRepository) Migrate(ctx context.Context) error {
	statements := []string{
`CREATE TABLE IF NOT EXISTS sales_reports (
    device_id TEXT NOT NULL,
    report_id TEXT NOT NULL,
    payload_hash BYTEA NOT NULL,
    occurred_at TIMESTAMPTZ NOT NULL,
    saved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (device_id, report_id)
)`,
`CREATE TABLE IF NOT EXISTS sales_report_lines (
    device_id TEXT NOT NULL,
    report_id TEXT NOT NULL,
    line_number INTEGER NOT NULL,
    sku TEXT NOT NULL,
    quantity BIGINT NOT NULL CHECK (quantity &gt; 0),
    amount_minor BIGINT NOT NULL CHECK (amount_minor &gt;= 0),
    PRIMARY KEY (device_id, report_id, line_number),
    FOREIGN KEY (device_id, report_id)
        REFERENCES sales_reports (device_id, report_id) ON DELETE CASCADE
	)`,
	}
	for _, statement := range statements {
		if _, err := r.db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("create sales schema: %w", err)
		}
	}
	return nil
}

func (r *SalesRepository) SaveReport(ctx context.Context, report contracts.SalesReport) (ReportOutcome, error) {
	if report.DeviceID == "" || report.ReportID == "" || len(report.Items) == 0 {
		return 0, fmt.Errorf("validated report identity and at least one item are required")
	}
	occurredAt, err := time.Parse(time.RFC3339, report.OccurredAt)
	if err != nil {
		return 0, fmt.Errorf("parse report time: %w", err)
	}
	encoded, err := json.Marshal(report)
	if err != nil {
		return 0, fmt.Errorf("encode report for deduplication: %w", err)
	}
	hash := sha256.Sum256(encoded)
	tx, err := r.db.BeginTx(ctx, &amp;sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return 0, fmt.Errorf("begin report transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }() // No effect after a successful commit.

	result, err := tx.ExecContext(ctx, `
INSERT INTO sales_reports (device_id, report_id, payload_hash, occurred_at)
VALUES ($1, $2, $3, $4)
ON CONFLICT (device_id, report_id) DO NOTHING`,
		report.DeviceID, report.ReportID, hash[:], occurredAt)
	if err != nil {
		return 0, fmt.Errorf("insert report identity: %w", err)
	}
	inserted, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("inspect report insert: %w", err)
	}
	if inserted == 0 {
		var storedHash []byte
		if err := tx.QueryRowContext(ctx, `
SELECT payload_hash FROM sales_reports WHERE device_id=$1 AND report_id=$2`,
			report.DeviceID, report.ReportID).Scan(&amp;storedHash); err != nil {
			return 0, fmt.Errorf("load duplicate report: %w", err)
		}
		if !equalBytes(storedHash, hash[:]) {
			return 0, ErrReportIDConflict
		}
		if err := tx.Commit(); err != nil {
			return 0, fmt.Errorf("commit duplicate observation: %w", err)
		}
		return ReportDuplicate, nil
	}
	for i, line := range report.Items {
		if _, err := tx.ExecContext(ctx, `
INSERT INTO sales_report_lines
    (device_id, report_id, line_number, sku, quantity, amount_minor)
VALUES ($1, $2, $3, $4, $5, $6)`,
			report.DeviceID, report.ReportID, i, line.SKU, line.Quantity, line.AmountMinor); err != nil {
			return 0, fmt.Errorf("insert sales line %d: %w", i, err)
		}
	}
	if err := tx.Commit(); err != nil {
		return 0, fmt.Errorf("commit sales report: %w", err)
	}
	return ReportSaved, nil
}

func equalBytes(left, right []byte) bool {
	if len(left) != len(right) {
		return false
	}
	var difference byte
	for i := range left {
		difference |= left[i] ^ right[i]
	}
	return difference == 0
}
</code></pre></details>
  <div class="card"><h3>🔎 Details: duplicate and transaction behavior</h3><p>The unique constraint handles concurrent transactions for one device/report pair. The winner inserts lines in the same transaction as the report row. A second identical report reads the stored digest and returns <code>ReportDuplicate</code>; different content under the same key returns <code>ErrReportIDConflict</code>. The deferred rollback is harmless after commit and cleans up early returns. Do not Ack the MQTT packet until this method returns a committed saved/duplicate outcome. This repository does not send the application response; publication intent belongs in an outbox when crash-safe receipts are required.</p></div>
  <h2>🧭 File overview: <code>internal/persistence/commands.go</code></h2>
  <p>Command state also needs durable storage. Create this complete file alongside <code>sales.go</code>. It stores validated device results under the stable command ID and allows a late result to resolve an earlier unknown outcome without losing the separate broker-ack field.</p>
  <h3>🏗️ <code>Migrate</code> — short summary</h3><p>Creates the command table and a unique result table.</p>
  <h3>📩 <code>ApplyResult</code> — short summary</h3><p>Inserts one result, accepts identical replays, detects conflicting IDs, and updates command status transactionally.</p>
  <details class="card" open><summary><strong>📄 <code>internal/persistence/commands.go</code> — complete file</strong></summary><pre><code>package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"

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

var ErrCommandResultConflict = errors.New("command ID has a conflicting result")

type CommandRepository struct { db *sql.DB }

func NewCommandRepository(db *sql.DB) (*CommandRepository, error) {
	if db == nil { return nil, fmt.Errorf("database connection is required") }
	return &amp;CommandRepository{db: db}, nil
}

func (r *CommandRepository) Migrate(ctx context.Context) error {
	statements := []string{
		`CREATE TABLE IF NOT EXISTS device_commands (
			command_id TEXT PRIMARY KEY, device_id TEXT NOT NULL, state TEXT NOT NULL,
			broker_acked BOOLEAN NOT NULL DEFAULT FALSE,
			deadline TIMESTAMPTZ,
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
		`CREATE TABLE IF NOT EXISTS command_results (
			command_id TEXT NOT NULL REFERENCES device_commands(command_id),
			device_id TEXT NOT NULL, message_id TEXT NOT NULL, status TEXT NOT NULL,
			detail TEXT NOT NULL DEFAULT '', occurred_at TIMESTAMPTZ NOT NULL,
			PRIMARY KEY(command_id,message_id), UNIQUE(device_id,message_id))`,
	}
	for _, statement := range statements {
		if _, err := r.db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("create command schema: %w", err)
		}
	}
	return nil
}

func (r *CommandRepository) Create(ctx context.Context, commandID, deviceID string) error {
	if commandID == "" || deviceID == "" { return fmt.Errorf("command ID and device ID are required") }
	_, err := r.db.ExecContext(ctx, `INSERT INTO device_commands(command_id,device_id,state)
VALUES ($1,$2,'pending')`, commandID, deviceID)
	if err != nil { return fmt.Errorf("create command: %w", err) }
	return nil
}

func (r *CommandRepository) MarkBrokerAck(ctx context.Context, commandID string) error {
	result, err := r.db.ExecContext(ctx, `UPDATE device_commands SET broker_acked=TRUE,updated_at=now()
WHERE command_id=$1`, commandID)
	if err != nil { return fmt.Errorf("record broker acknowledgement: %w", err) }
	rows, err := result.RowsAffected()
	if err != nil { return fmt.Errorf("inspect command update: %w", err) }
	if rows != 1 { return fmt.Errorf("command %q not found", commandID) }
	return nil
}

func (r *CommandRepository) ApplyResult(ctx context.Context, result contracts.CommandResult) error {
	if result.CommandID == "" || result.DeviceID == "" || result.MessageID == "" {
		return fmt.Errorf("validated result identity is required")
	}
	occurredAt, err := time.Parse(time.RFC3339, result.OccurredAt)
	if err != nil { return fmt.Errorf("parse result time: %w", err) }
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil { return fmt.Errorf("begin result transaction: %w", err) }
	defer func() { _ = tx.Rollback() }()
	insert, err := tx.ExecContext(ctx, `INSERT INTO command_results
(command_id,device_id,message_id,status,detail,occurred_at)
VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (command_id,message_id) DO NOTHING`,
		result.CommandID,result.DeviceID,result.MessageID,result.Status,result.Detail,occurredAt)
	if err != nil { return fmt.Errorf("insert command result: %w", err) }
	rows, err := insert.RowsAffected()
	if err != nil { return fmt.Errorf("inspect result insert: %w", err) }
	if rows == 0 {
		var oldDevice, oldMessage, oldStatus, oldDetail string
		if err := tx.QueryRowContext(ctx, `SELECT device_id,message_id,status,detail
FROM command_results WHERE command_id=$1 AND message_id=$2`, result.CommandID,result.MessageID).Scan(&amp;oldDevice,&amp;oldMessage,&amp;oldStatus,&amp;oldDetail); err != nil {
			return fmt.Errorf("load prior command result: %w", err)
		}
		if oldDevice != result.DeviceID || oldMessage != result.MessageID || oldStatus != result.Status || oldDetail != result.Detail {
			return ErrCommandResultConflict
		}
		return tx.Commit() // Identical replay: keep the prior terminal state.
	}
	updated, err := tx.ExecContext(ctx, `UPDATE device_commands SET state=$2,updated_at=now()
WHERE command_id=$1 AND device_id=$3 AND state IN ('pending','accepted','unknown')`,
		result.CommandID,result.Status,result.DeviceID)
	if err != nil { return fmt.Errorf("update command state: %w", err) }
	changed, err := updated.RowsAffected()
	if err != nil { return fmt.Errorf("inspect command state update: %w", err) }
	if changed != 1 { return fmt.Errorf("command missing, device mismatch, or terminal state conflict") }
	if err := tx.Commit(); err != nil { return fmt.Errorf("commit command result: %w", err) }
	return nil
}
</code></pre></details>
  <div class="card"><h3>🔎 Details: legal state transitions</h3><p>The <code>state IN ('pending','accepted','unknown')</code> guard prevents a new result from overwriting an already terminal state such as completed or rejected. The result row and command-state update share a transaction. <code>broker_acked</code> remains a separate transport fact. Before using this example, ensure the command row exists and the result's topic/body identity was validated in the ingestion service.</p></div>
  <h2>🧹 Invalid and transient failures need different paths</h2>
  <table><thead><tr><th>Outcome</th><th>Durable action</th><th>MQTT decision</th></tr></thead><tbody><tr><td>Valid new report</td><td>Commit identity and effect</td><td>Ack after commit</td></tr><tr><td>Valid duplicate</td><td>Load prior result; no repeated effect</td><td>Ack after recognizing it</td></tr><tr><td>Permanently invalid</td><td>Commit bounded quarantine/rejection metadata</td><td>Ack after disposition commits</td></tr><tr><td>Transient database failure</td><td>Do not claim durable acceptance</td><td>Do not Ack; invoke explicit recovery/backoff</td></tr></tbody></table>
  <p>A poison message must not occupy a Receive Maximum slot forever. Record safe metadata such as topic, reason category, and receipt time; avoid secrets and arbitrary payload dumps. A transient database outage is not permanent invalid input.</p>
  <h2>🧪 Crash-point experiments</h2>
  <ol><li>Send the same report twice sequentially and concurrently. Expect one durable business effect.</li><li>Force a transaction rollback and verify neither dedupe identity nor business effect commits alone.</li><li>Commit, stop the process before manual Ack, reconnect with an eligible broker session, and observe a safe duplicate.</li><li>Send a permanently invalid report and confirm a bounded rejection record commits before Ack.</li><li>Delay acknowledgements to the configured Receive Maximum and observe delivery pressure. Remember this protocol limit does not cap your Go worker queue.</li><li>Record the gap between database commit and publishing an application response; Lesson 14 closes that gap with an outbox.</li></ol>
  <div class="card"><h3>🧱 Keep durable and temporary state distinct</h3><p>Persist reports, deduplication keys, command state/results, and any publication intent needed after restart. Keep live connections, worker queues, in-flight Paho packet handles, and process contexts in memory. On restart, recover from durable rows and eligible MQTT sessions; a dead callback cannot resume.</p><p><strong>Evidence boundary:</strong> database COMMIT establishes the backend's durable acceptance policy. PUBACK completes the MQTT protocol delivery. A separate application receipt tells the device the backend outcome, and it can still be lost unless its publication intent is persisted.</p></div>
  <h2>✅ Done when</h2><p>After a backend restart, replay does not double-count a report; a database outage is not acknowledged as durable success; invalid data has a durable disposition; and you can explain why manual Ack alone is not permanent broker storage.</p>
  <p>Run only against the dedicated lab PostgreSQL service and synthetic data. The full source lesson includes the project-specific Docker commands and Paho acknowledgement references.</p>
  <details class="card"><summary><strong>📋 Short self-check answers</strong></summary><ol><li>Unique stable business key plus one transaction for identity/effect/outcome.</li><li>Commit then Ack; after crash-before-Ack, recognize duplicate and avoid repeating the effect.</li><li>Callback error alone does not request retry; recovery is an application policy.</li><li>Manual Ack is not a durable application queue and depends on session/broker behavior.</li></ol></details>
  <p>📚 Failure experiments: <a href="../course-source/13-persist-reports-command-state.md">Lesson 13 source lesson</a>. Primary references: <a href="https://pkg.go.dev/database/sql">Go <code>database/sql</code></a>, <a href="https://www.postgresql.org/docs/current/ddl-constraints.html">PostgreSQL constraints</a>, and the <a href="https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html">OASIS MQTT 5.0 Standard</a>.</p>
  <p>Ask me to reason through the exact crash point you observed before changing the acknowledgement policy.</p>
  <footer><p>Previous: <a href="0012-vending-workflows.html">Lesson 12 — vending workflows</a> · Next: <a href="0014-outbox-failure-gaps.html">Lesson 14 — transactional outbox</a> · Detailed source: <a href="../course-source/14-database-mqtt-failure-gaps.md">Markdown lesson</a>.</p></footer>
</main></body></html>