skaidb v0.234.0

MQTT broker

skaidb nodes speak MQTT natively: Home Assistant, zigbee2mqtt, Tasmota, ESPHome, Shelly and any standard MQTT client connect directly to the database — no separate Mosquitto in the stack. Both MQTT 3.1.1 and 5.0 are served (plus read-compat MQTT 3.1), over TCP, TLS, and WebSocket.

What makes it different from a file-backed broker: broker state lives in replicated skaidb tables (the reserved _mqtt database). Retained messages, persistent sessions, subscriptions and offline queues survive a node restart — and on a cluster they survive node loss: a client can disconnect from one node and resume its session, with its queued messages, on any other node. Broker state is also SQL-inspectable:

SELECT topic, stored_at FROM _mqtt.retained WHERE topic LIKE 'homeassistant/%';
SELECT client_id, disconnected_at FROM _mqtt.sessions;

Enabling

[mqtt]
enabled = true
port = 1883                  # plain / opportunistic TLS per [encryption].client_tls
tls_port = 8883              # always-TLS; bound only when client_tls != "off"
allow_anonymous = false      # see Authentication below

enabled and the ports need a restart; every limit is runtime-mutable via config set. The full knob list with defaults is in config/skaidb.toml. CLI/env overrides: --mqtt-enabled / SKAIDB_MQTT_ENABLED, --mqtt-port, --mqtt-tls-port, --mqtt-allow-anonymous.

Transports

  • TCP on mqtt.port (1883). With client_tls = "opportunistic" the same port also accepts TLS; with "required", plaintext is refused.
  • TLS on mqtt.tls_port (8883), always-TLS, using the [encryption] certificate.
  • WebSocket (ws:// and wss://): auto-detected on the same listeners — an HTTP GET upgrade (RFC 6455, subprotocol mqtt) instead of an MQTT CONNECT. No extra port or configuration.

Authentication & authorization

CONNECT username/password authenticates against the same user catalog as every other endpoint (CREATE USER … PASSWORD …). Failures answer CONNACK 0x05/0x04 (3.1.1) or 0x86 (5.0).

mqtt.allow_anonymous is an extra gate on top of [auth]: even when server auth is disabled, the MQTT listener refuses credential-less CONNECTs unless it is true — the IoT listener is usually the most exposed surface on the box. Per-IP CONNECT rate limiting (mqtt.connect_rate_limit, default 20/s with 2× burst) blunts reconnect storms.

Topic ACLs

By default any authenticated user may publish/subscribe anywhere (what a default Mosquitto gives Home Assistant). Roles opt into topic ACLs via grants; a role with any topic grant is restricted to its grants, and the superuser is never restricted:

GRANT PUBLISH   ON TOPIC 'tele/#'     TO sensors;
GRANT SUBSCRIBE ON TOPIC 'cmnd/+/set' TO sensors;
REVOKE PUBLISH  ON TOPIC 'tele/#'     FROM sensors;

Grant filters use MQTT wildcard semantics. A publish is allowed when any granted filter matches the topic; a subscribe is allowed when any granted filter covers the requested filter (a grant of home/# covers a subscription to home/+/temp). Denied publishes answer 0x87 Not authorized on 5.0 (3.1.1 has no error channel: the packet is acknowledged per QoS and silently not routed, counted in skaidb_mqtt_messages_dropped_total{reason="acl"}); denied subscriptions get a per-filter SUBACK failure. A will whose topic the role may not publish is refused at CONNECT. Grant changes apply on the client's next connect.

Sessions & QoS

Full QoS 0/1/2 in both directions. Persistent sessions (3.1.1 CleanSession=0, 5.0 Session Expiry Interval > 0) keep their subscriptions, unacknowledged deliveries and offline queue in _mqtt.* tables:

  • QoS 1/2 publishes to a persistent session are written to _mqtt.queue before the publisher's ack — the ack is the durability promise. QoS 0 to a live subscriber never touches storage.
  • Resumption replays the backlog in order: unacknowledged legs are retransmitted with DUP=1 under their original packet ids, and the QoS 2 PUBREL leg resumes as PUBREL (recorded durably before it is first sent) — exactly-once holds across reconnects and broker restarts.
  • Offline queues are bounded (mqtt.max_queued_per_session); sessions expire after mqtt.session_expiry_max_secs offline (5.0 clients may request less, and may revise it at DISCONNECT).
  • A second CONNECT with the same client id takes the session over and disconnects the first (5.0 reason 0x8E), on whichever node it lives.

Retained messages are cached in memory and persisted write-behind (~100 ms batches) to _mqtt.retained — a crash may lose the last ≤100 ms of retained churn, matching ecosystem practice. Wills fire on every abnormal disconnect (never on clean DISCONNECT; 5.0 reason 0x04 keeps the will on a clean close), honor the 5.0 Will Delay Interval, cancel on reconnect, and survive broker restarts.

MQTT 5.0 specifics

Server capabilities are advertised in CONNACK (Receive Maximum, Topic Alias Maximum 64, Maximum Packet Size, Assigned Client Identifier, capped Session Expiry). Message Expiry is honored end-to-end (queued copies expire in place; deliveries carry the remaining interval; retained messages expire). Subscription options (No Local, Retain As Published, Retain Handling), subscription identifiers, inbound topic aliases, request/response and user-property pass-through, and shared subscriptions ($share/{group}/{filter}, round-robin preferring live members) are all supported. Egress honors the client's Maximum Packet Size by dropping, never truncating.

Clustering

Enable [mqtt] on every member; clients may connect to any node. Publishes fan out to peers over the internode transport and each node delivers to its own connected clients; sessions resume on any node (replicated state). QoS 1/2 to a subscriber on another node — or on a node that has died — rides _mqtt.queue, so acknowledged messages are never lost to a node failure. Cluster notes:

  • Shared-subscription round-robin is cluster-approximate (the origin node picks the member); members that must receive across nodes should use persistent sessions.
  • $SYS topics are per-node.
  • _mqtt.* tables default to witness = false (transient operational state; witnesses skip them).
  • Cross-node QoS 2 degrades to effectively-once at the internode hop (documented, per the RFC; the queue-row dedup makes duplicates rare).

Topic → table capture (the sink)

Config-driven rules capture matching publishes into tables — many topics to one table, with wildcard captures as column/label values:

[[mqtt.sink]]
filter        = "home/+/+/state"      # + captures: room, device
table         = "iot.sensor_state"
mode          = "row"                 # JSON payload fields → columns
topic_columns = ["room", "device"]

[[mqtt.sink]]
filter = "zigbee2mqtt/+/SENSOR"
table  = "iot.metrics"
mode   = "timeseries"                 # numeric JSON leaves → samples
series = ["device"]                   # capture label names

row mode inserts one row per message: JSON object fields become columns (non-JSON payloads land in a payload bytes column), plus the captures, topic, and ts. timeseries mode flattens numeric JSON leaves into the remote_write fast path (each leaf a series named by its dotted path, string fields and captures as labels), auto-creating the TS table — MQTT telemetry becomes PromQL-queryable with zero glue services (see TIMESERIES.md / GRAFANA.md).

The sink enforces the publishing user's Insert privilege on the target table and the read-only gate, exactly like remote_write. With ack_on_sink = true the publisher's ack is gated on the table write (5.0 answers 0x97 on failure — a durable ingest API). Out-of-order-window TS drops are counted (skaidb_mqtt_sink_dropped_total{reason="ooo"}), not fatal. Sink rules are runtime-mutable.

$SYS topics

A Mosquitto-compatible subset is published every 10 s per node (mqtt.sys_topics_enabled, default on): version, uptime, clients connected/total, messages received/sent, subscription and retained counts. $SYS values are retained in cache only and are never matched by a plain # subscription (per spec).

The $ tree is broker-owned: a PUBLISH from a client to any $-prefixed topic is refused and the connection dropped, whatever its ACL allows and including a superuser, so nothing can pin attacker-chosen values into the retained cache and feed them to whatever monitors $SYS. The stats publisher above writes through an internal path that never passes this check.

Operational notes

  • MQTT connections appear in the drivers connections registry (endpoint mqtt) and the UI.
  • Metrics: skaidb_connections_{total,active}{endpoint="mqtt"}, skaidb_mqtt_packets_total{type,dir}, skaidb_mqtt_connect_total{outcome}, skaidb_mqtt_messages_dropped_total{reason}, skaidb_mqtt_sink_* — see METRICS.md.
  • Broker-state writes use mqtt.state_consistency (default quorum).
  • Graceful shutdown DISCONNECTs clients (5.0 reason 0x8B) and flushes dirty broker state; wills are not published on server shutdown.
  • Every limit is enforced from the first byte (max packet size checked before a body is buffered; bounded outboxes, queues, retained store, topic shape, subscriptions per session).
  • Deliberate deferrals: GSSAPI over 5.0 AUTH, TLS client-certificate identity, MQTT-SN/QUIC, broker bridging, publish-from-SQL.

Home Assistant quick start

Point the HA mqtt integration (or zigbee2mqtt's mqtt.server) at mqtt://<node>:1883 with a database user's credentials. Discovery, retained state, availability (LWT), and QoS 0/1/2 work out of the box; after a broker restart HA re-reads its retained discovery topics from skaidb's replicated store.