Deep Dive: How MQTT 5.0 Implements QoS Levels for Reliable IoT Message Delivery

import paho.mqtt.client as mqtt
import time
import logging
from typing import Dict, Optional

# Configure logging for production visibility
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class QoS1Publisher:
    \"\"\"Production-ready MQTT 5.0 QoS 1 publisher with retry logic and metrics.\"\"\"

    def __init__(self, broker_host: str, broker_port: int = 1883, client_id: str = "qos1-publisher"):
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.client_id = client_id
        self.client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv5)
        self.pending_acks: Dict[int, float] = {}  # packet_id -> publish timestamp
        self.publish_count = 0
        self.ack_timeout = 5.0  # 5 second timeout for PUBACK

        # Register callbacks
        self.client.on_connect = self._on_connect
        self.client.on_publish = self._on_publish
        self.client.on_disconnect = self._on_disconnect

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_success:
            logger.info(f"Connected to broker {self.broker_host}:{self.broker_port}")
        else:
            logger.error(f"Connection failed: {reason_code.getName()}")

    def _on_publish(self, client, userdata, mid):
        \"\"\"Callback for PUBACK received (QoS 1) or PUBREC (QoS 2)\"\"\"
        if mid in self.pending_acks:
            latency = time.time() - self.pending_acks[mid]
            logger.info(f"PUBACK received for packet {mid}, latency: {latency:.3f}s")
            del self.pending_acks[mid]
        else:
            logger.warning(f"Received ACK for unknown packet ID {mid}")

    def _on_disconnect(self, client, userdata, reason_code, properties):
        logger.warning(f"Disconnected: {reason_code.getName()}. Reconnecting...")
        time.sleep(1)
        self.client.reconnect()

    def publish(self, topic: str, payload: bytes, qos: int = 1) -> None:
        \"\"\"Publish a message with QoS 1, track pending ACKs, handle timeouts.\"\"\"
        if qos != 1:
            raise ValueError("This publisher only supports QoS 1")
        try:
            # Publish returns the message ID (mid) for QoS > 0
            msg_info = self.client.publish(topic, payload, qos=qos, retain=False)
            if msg_info.rc != mqtt.MQTT_ERR_SUCCESS:
                raise RuntimeError(f"Publish failed: {mqtt.error_string(msg_info.rc)}")
            self.pending_acks[msg_info.mid] = time.time()
            self.publish_count += 1
            logger.debug(f"Published message {msg_info.mid} to {topic}")
        except Exception as e:
            logger.error(f"Publish error: {str(e)}")
            raise

    def check_timeouts(self) -> None:
        \"\"\"Periodically check for pending ACKs that have timed out.\"\"\"
        current_time = time.time()
        timed_out = [mid for mid, ts in self.pending_acks.items() if current_time - ts > self.ack_timeout]
        for mid in timed_out:
            logger.error(f"Packet {mid} timed out after {self.ack_timeout}s, resending...")
            # In production, you'd resend the original payload here; for brevity we log
            del self.pending_acks[mid]

    def run(self, topic: str, payload: bytes, interval: float = 1.0) -> None:
        \"\"\"Main loop: publish periodically, check timeouts.\"\"\"
        try:
            self.client.connect(self.broker_host, self.broker_port)
            self.client.loop_start()
            while True:
                self.publish(topic, payload)
                self.check_timeouts()
                time.sleep(interval)
        except KeyboardInterrupt:
            logger.info("Shutting down...")
        finally:
            self.client.loop_stop()
            self.client.disconnect()
            logger.info(f"Total published: {self.publish_count}")

if __name__ == "__main__":
    # Example usage: publish to test topic every 1 second
    publisher = QoS1Publisher(broker_host="mqtt-broker.example.com", broker_port=1883)
    publisher.run(topic="iot/sensors/temp", payload=b'{"temp": 22.5, "ts": 1717234567}', interval=1.0)

import com.hivemq.extension.sdk.api.annotations.NotNull;
import com.hivemq.extension.sdk.api.packets.publish.PublishPacket;
import com.hivemq.extension.sdk.api.packets.publish.Qos;
import com.hivemq.extension.sdk.api.services.exception.RateLimitException;
import com.hivemq.extension.sdk.api.services.publish.PublishService;
import com.hivemq.extension.sdk.api.services.subscription.SubscriptionStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * HiveMQ Extension to enforce QoS 2 state persistence for inbound publishes.
 * Implements the MQTT 5.0 QoS 2 four-packet exchange state machine.
 */
public class QoS2StateManager {
    private static final @NotNull Logger logger = LoggerFactory.getLogger(QoS2StateManager.class);
    private static final int MAX_PENDING_QOS2 = 10_000;  // Max pending QoS 2 packets per broker

    // State map: packetId -> QoS2State (scoped to client ID + packet ID)
    private final @NotNull ConcurrentMap pendingStates = new ConcurrentHashMap<>();
    private final @NotNull PublishService publishService;
    private final @NotNull SubscriptionStore subscriptionStore;

    public QoS2StateManager(final @NotNull PublishService publishService, 
                           final @NotNull SubscriptionStore subscriptionStore) {
        this.publishService = publishService;
        this.subscriptionStore = subscriptionStore;
    }

    /**
     * Handle inbound QoS 2 PUBLISH packet from a client.
     * State 1: Received PUBLISH, store state, send PUBREC.
     */
    public void handleInboundPublish(final @NotNull String clientId, 
                                    final int packetId, 
                                    final @NotNull PublishPacket packet) {
        if (packet.getQos() != Qos.EXACTLY_ONCE) {
            logger.debug("Ignoring non-QoS 2 packet: {}", packet.getQos());
            return;
        }
        if (pendingStates.size() >= MAX_PENDING_QOS2) {
            logger.error("Max pending QoS 2 states reached, dropping packet {}", packetId);
            // Send DISCONNECT with Server Unavailable reason code
            return;
        }
        final String stateKey = clientId + "-" + packetId;
        final QoS2State existing = pendingStates.get(stateKey);
        if (existing != null) {
            logger.warn("Duplicate QoS 2 PUBLISH for existing state: {}", stateKey);
            return;  // Already processed, ignore duplicate
        }
        // Persist state to disk in production; here we use in-memory for brevity
        final QoS2State newState = new QoS2State(clientId, packetId, packet, System.currentTimeMillis());
        pendingStates.put(stateKey, newState);
        logger.info("Stored QoS 2 state for {}. Sending PUBREC.", stateKey);
        // In actual HiveMQ extension, you'd send PUBREC via the client connection here
    }

    /**
     * Handle PUBREL packet from client (State 2: Received PUBREL, delete state, send PUBCOMP).
     */
    public void handlePubRel(final @NotNull String clientId, final int packetId) {
        final String stateKey = clientId + "-" + packetId;
        final QoS2State state = pendingStates.get(stateKey);
        if (state == null) {
            logger.warn("No pending QoS 2 state for PUBREL: {}", stateKey);
            return;
        }
        // Forward the original PUBLISH to subscribers
        try {
            publishService.publish(state.getPacket()).get();  // Blocking for simplicity
            logger.info("Forwarded QoS 2 packet {} to subscribers", stateKey);
        } catch (final RateLimitException e) {
            logger.error("Rate limited while forwarding QoS 2 packet: {}", stateKey, e);
        } catch (final Exception e) {
            logger.error("Failed to forward QoS 2 packet: {}", stateKey, e);
        } finally {
            pendingStates.remove(stateKey);
            logger.info("Removed QoS 2 state for {}. Sending PUBCOMP.", stateKey);
            // Send PUBCOMP to client here
        }
    }

    /**
     * Clean up expired states (older than 1 hour) to prevent memory leaks.
     */
    public void cleanupExpiredStates() {
        final long cutoff = System.currentTimeMillis() - 3_600_000;  // 1 hour
        pendingStates.entrySet().removeIf(entry -> entry.getValue().getTimestamp() < cutoff);
        logger.debug("Cleaned up expired QoS 2 states. Pending count: {}", pendingStates.size());
    }

    /**
     * Internal state class for QoS 2 pending packets.
     */
    private static class QoS2State {
        private final @NotNull String clientId;
        private final int packetId;
        private final @NotNull PublishPacket packet;
        private final long timestamp;

        QoS2State(final @NotNull String clientId, final int packetId, 
                  final @NotNull PublishPacket packet, final long timestamp) {
            this.clientId = clientId;
            this.packetId = packetId;
            this.packet = packet;
            this.timestamp = timestamp;
        }

        @NotNull PublishPacket getPacket() { return packet; }
        long getTimestamp() { return timestamp; }
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

const (
    brokerURL      = "tcp://mqtt-broker.example.com:1883"
    clientID       = "qos2-subscriber"
    subscribeTopic = "iot/control/+"
    qosLevel       = 2  // Exactly once
    ackTimeout     = 5 * time.Second
)

// QoS2Subscriber manages a MQTT 5.0 QoS 2 subscriber with state tracking
type QoS2Subscriber struct {
    client         mqtt.Client
    opts           *mqtt.ClientOptions
    pendingPUBREC  map[uint16]time.Time  // packet ID -> PUBREC receive time
    recvCount      uint64
    ackCount       uint64
}

// NewQoS2Subscriber initializes a new QoS 2 subscriber
func NewQoS2Subscriber() *QoS2Subscriber {
    opts := mqtt.NewClientOptions()
    opts.AddBroker(brokerURL)
    opts.SetClientID(clientID)
    opts.SetProtocolVersion(5)  // MQTT 5.0
    opts.SetAutoReconnect(true)
    opts.SetMaxReconnectInterval(30 * time.Second)
    opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
        log.Printf("Connection lost: %v", err)
    })
    opts.SetOnConnectHandler(func(client mqtt.Client) {
        log.Println("Connected to broker")
    })

    return &QoS2Subscriber{
        opts:          opts,
        pendingPUBREC: make(map[uint16]time.Time),
    }
}

// start begins the subscription loop
func (s *QoS2Subscriber) start(ctx context.Context) error {
    s.client = mqtt.NewClient(s.opts)
    if token := s.client.Connect(); token.Wait() && token.Error() != nil {
        return fmt.Errorf("failed to connect: %w", token.Error())
    }
    defer s.client.Disconnect(250)

    // Subscribe to topic with QoS 2
    subscribeToken := s.client.Subscribe(subscribeTopic, byte(qosLevel), s.messageHandler)
    if subscribeToken.Wait() && subscribeToken.Error() != nil {
        return fmt.Errorf("failed to subscribe: %w", subscribeToken.Error())
    }
    log.Printf("Subscribed to %s with QoS %d", subscribeTopic, qosLevel)

    // Start timeout checker goroutine
    go s.checkTimeouts(ctx)

    // Wait for shutdown signal
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    <-sigChan
    log.Println("Shutting down...")
    return nil
}

// messageHandler processes incoming QoS 2 PUBLISH packets
func (s *QoS2Subscriber) messageHandler(client mqtt.Client, msg mqtt.Message) {
    packetID := msg.PacketID()
    log.Printf("Received QoS 2 message %d on topic %s: %s", packetID, msg.Topic(), string(msg.Payload()))

    // For QoS 2, we need to send PUBREC, wait for PUBREL, send PUBCOMP
    // The paho library handles this internally, but we track state for metrics
    s.pendingPUBREC[packetID] = time.Now()
    s.recvCount++

    // Simulate processing time (e.g., writing to DB)
    time.Sleep(100 * time.Millisecond)

    // Acknowledge the message (PUBREC is sent automatically by paho for QoS 2)
    msg.Ack()
    s.ackCount++
    log.Printf("Acknowledged message %d. Total received: %d, ACKed: %d", packetID, s.recvCount, s.ackCount)
}

// checkTimeouts periodically checks for pending PUBREC that have timed out
func (s *QoS2Subscriber) checkTimeouts(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            now := time.Now()
            for packetID, ts := range s.pendingPUBREC {
                if now.Sub(ts) > ackTimeout {
                    log.Printf("Packet %d timed out after %v, resending PUBREC?", packetID, ackTimeout)
                    // In production, resend PUBREC here; paho handles internally
                    delete(s.pendingPUBREC, packetID)
                }
            }
        }
    }
}

func main() {
    subscriber := NewQoS2Subscriber()
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    if err := subscriber.start(ctx); err != nil {
        log.Fatalf("Subscriber failed: %v", err)
    }
}

def on_subscribe(client, userdata, mid, granted_qos, properties):
    for topic, qos in zip(userdata["subscribe_topics"], granted_qos):
        if qos < userdata["requested_qos"]:
            logger.warning(f"Broker downgraded QoS for {topic} to {qos} from {userdata['requested_qos']}")
        else:
            logger.info(f"Subscribed to {topic} with QoS {qos}")

# Mosquitto config snippet for QoS 2 persistence
persistence true
persistence_location /var/lib/mosquitto/
persistence_file mosquitto.db
# Set QoS 2 state timeout to 1 hour
qos2_state_timeout 3600

# Python paho-mqtt client timeout tuning
client = mqtt.Client(protocol=mqtt.MQTTv5)
client._transport_timeout = 3.0  # 3 second transport timeout
# For QoS 1, check pending ACKs every second
def check_qos_timeouts():
    for mid, ts in pending_acks.items():
        if time.time() - ts > 3.0:  # 3s timeout for CAT-M1
            resend_message(mid)