skaidb v0.263.1

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. The MQTT listeners present the [encryption] certificate unless mqtt.tls_cert_file / mqtt.tls_key_file name their own pair (a public device endpoint on its own hostname while the SQL ports keep the internal one); either pair rotates live when its files change. The MQTT acceptor offers ALPN mqtt; clients that send no ALPN still connect.
  • 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).

Two ways in that never put a password on the wire, tried before the username/password fields:

  • 5.0 enhanced authentication (AUTH packets). A CONNECT carrying an Authentication Method runs the exchange of §4.12 — AUTH 0x18 both ways, the CONNACK closes it with the method and the server's final message in its properties. Methods offered:
  • SCRAM-SHA-256, RFC 5802 message syntax, so any SCRAM client library produces the right bytes: CONNECT Authentication Data = n,,n=<user>,r=<nonce> (=2C/=3D escaping in the user); the server AUTHs r=<nonce+ext>,s=<base64 salt>,i=<iterations>; the client AUTHs c=biws,r=<nonce+ext>,p=<base64 proof>; the CONNACK's data is v=<base64 server signature> — verify it, the server proved itself too. The SCRAM user is the identity (the CONNECT username/password fields are ignored). A failed proof is 0x86; stepping out of the exchange (any packet but AUTH 0x18) is 0x82.
  • GSSAPI (Kerberos), on a kerberos build with auth.gssapi_enabled: CONNECT data = the initial token, AUTH packets shuttle the rest, the CONNACK carries the final mutual-auth token; the principal maps to an external user (CREATE USER "<principal>" GSSAPI) exactly as on the binary port. Not authorized is 0x87.
  • anything else, or a method this build/config does not offer: 0x8C (Bad authentication method). Re-authentication (AUTH 0x19 mid-session) is not supported; a session that wants fresh credentials reconnects.
  • TLS client-certificate identity (auth.x509_enabled = true, auth.x509_ca_file = <CA PEM>). Every client-TLS port then requests a client certificate (a client without one still connects and authenticates otherwise). On the MQTT listener a certificate chaining to that CA is the credential: its subject Common Name is the username (the Mosquitto use_identity_as_username convention) — no password; the user must exist (CREATE USER "device-7" …, any kind). A CONNECT username that disagrees with the CN is refused (0x05/0x87); a CN naming no user is 0x04/0x86. Certificate rotation is live like the server's own (the acceptor rebuilds when the PEM files change).

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.

Bridging to another broker

[[mqtt.bridge]] mirrors topics between this broker and another one — the migration or federation tool Mosquitto users know as a bridge connection:

[[mqtt.bridge]]
name = "cloud"
remote = "broker.example.com:8883"
remote_tls = true
remote_tls_ca = "/etc/skaidb/cloud-ca.pem"   # empty = the public roots
remote_user = "site42"
remote_password = "…"
local_user = "bridge"        # a catalog user holding the topic grants
local_password = "…"
out = ["tele/#", "state/#"]  # local → remote as <remote_prefix><topic>
remote_prefix = "site42/"
in = ["site42/cmd/#"]        # remote → local as <local_prefix><topic>
local_prefix = ""
qos = 1
keep_alive_secs = 30

A bridge is a thread holding two ordinary MQTT 3.1.1 clients: a loopback link into this broker (so it is subject to local_user's topic grants, admission limits and QoS like any device) and a link to the remote. A message a bridge brings in is a client publish here — forwarded to cluster peers, captured by sinks, queued for offline sessions. Rules: out and in must not overlap after remapping (MQTT carries no origin, so that is the only loop prevention, exactly as in Mosquitto); retain does not cross a bridge (a 3.1.1 subscriber never sees a publisher's retain flag on a live delivery); either link failing drops both, and the bridge reconnects with a capped backoff (1 s to 30 s) and resubscribes — messages during the outage are not replayed. Counters: skaidb_mqtt_bridge_messages_total{bridge,direction} and skaidb_mqtt_bridge_reconnects_total{bridge}. With client_tls = "required" the loopback link uses the TLS port and trusts the node's own certificate.

Publishing from SQL

PUBLISH '<topic>' PAYLOAD <expr> [QOS 0|1|2] [RETAIN] sends one message through this node's broker — subscribers here get it, peers forward it like any client publish, RETAIN stores it for late subscribers. A string payload publishes its bytes, a document or array its JSON:

PUBLISH 'tele/kitchen' PAYLOAD {"t": 21.5, "unit": 'C'} QOS 1;
PUBLISH 'state/door' PAYLOAD 'open' RETAIN;

The topic is a literal (concrete name: no wildcards, not $), so the check is static: the role needs PUBLISH on it under the same topic grants a client connection gets (GRANT PUBLISH ON TOPIC 'tele/#' TO role; a role with no topic grants keeps the open default). A procedure body may PUBLISH, which is how a trigger turns a row change into an MQTT message (change data capture without a separate stream tail):

CREATE PROCEDURE announce(id TEXT, total INT)
BEGIN
  PUBLISH 'orders/new' PAYLOAD {"id": id, "total": total};
END;
CREATE TRIGGER order_out ON orders WHEN (total > 0) CALL announce(NEW.id, NEW.total);

The trigger runs as its definer; a CALL by hand is gated on the body's topics. Delivery only — topic→table capture rules ([[mqtt.sink]]) run on client publishes, under the publishing client's role, not on SQL ones. Needs the broker ([mqtt] enabled = true); otherwise the statement says so.

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). They are ordinary replicated tables otherwise, so BACKUP TO / BACKUP CLUSTER TO include them — retained messages, persistent sessions, subscriptions and offline queues restore with the rest of the node.
  • 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.
  • Idle parking (epoll). A connection is served by a reader thread and a writer thread while it is talking; once it has been quiet for mqtt.idle_park_secs (default 2), its socket moves to the broker's epoll reactor and both threads exit. The reactor resumes it on the next byte (a ping, a publish) on a fresh thread — the same loop, same session — and closes it at its keep-alive deadline if nothing arrives; the writer is respawned by the next message queued for it. Ten thousand idle devices therefore cost ten thousand registered descriptors, not threads; active connections keep the thread-per-connection path with its blocking, simple code. idle_park_secs = 0 disables parking; on Windows there is no epoll reactor and every connection keeps its thread pair. Counters: skaidb_mqtt_parked_readers (gauge), skaidb_mqtt_parks_total{side}, skaidb_mqtt_wakes_total{side}.
  • 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: MQTT-SN/QUIC, AUTH re-authentication, MQTT 5 across bridges (retain-as-published).

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.