# MQTT v5 server side sessions and routing
MQTT v5 session and topic routing - embeddable in Erlang projects.
This library handles pools of MQTT sessions.
Transports (connections) attach to MQTT sessions to relay packets.
The sessions handles packets, queues, and a user-context state.
Access control and authentication is handled with a runtime module.
This module is configured in the `mqtt_sessions` application env key `runtime`.
The default and example runtime is `src/mqtt_sessions_runtime.erl`.
Note that this library does not handle TCP/IP connections. It handles the
complete MQTT session logic. Other libraries are used for transporting the
MQTT packets to/from external clients.
When a subscription is made to a $SYS topic the subscriber is mapped to
the default pool. This makes it possible to share system information
between different pools.
## How connects work
`mqtt_sessions` keeps the MQTT session state separate from the transport that
carries packets.
An incoming MQTT `CONNECT` does not always create a brand new session. It can:
- create a new session for a new `client_id`
- reattach to an existing session for the same `client_id`
- replace the currently attached transport for that session
This means that the same MQTT session can survive disconnects and later be
reattached by another connection source, as long as the session has not expired
and the protocol flags allow the reconnect.
The attached transport can come from different sources. For example, one
session can be driven by a websocket bridge used by a browser, while another
session can be driven by a native MQTT listener connection. The session process
only manages MQTT state; the actual transport process is external and can vary
per reconnect.
## Disconnect timeout behaviour
Sessions normally keep the negotiated `session_expiry_interval` after a disconnect.
Runtimes can mark a session as:
- websocket-originated
- anonymous
- crawler
When a session is both websocket-originated and anonymous, `mqtt_sessions`
uses a shorter timeout after the first disconnect:
- anonymous websocket session: 30 seconds
- anonymous websocket crawler session: 10 seconds
This is intended to reduce memory use for one-page anonymous visits, such as
headless crawler traffic that creates many short-lived browser sessions without
reusing them.
Authenticated websocket sessions and non-websocket sessions keep their normal
session expiry behaviour.
## Runtime callbacks
The runtime module is responsible for authentication, authorization, and for
describing some session characteristics derived from the opaque `user_context()`.
The runtime callbacks are:
- `vhost_pool/1`
Map a virtual host to a session pool.
- `pool_default/0`
Return the default pool when no host-specific pool can be resolved.
- `new_user_context/3`
Create the initial runtime-specific `user_context()` for a session.
- `connect/4`
Handle MQTT `CONNECT` and return `CONNACK` plus an updated user context.
- `reauth/2`
Handle MQTT v5 re-authentication.
- `is_allowed/4`
Check if a publish or subscribe action is allowed for a topic.
- `is_valid_message/3`
Validate an incoming message before it is processed.
- `control_message/3`
Handle runtime-specific control messages and update the user context.
- `is_websocket_origin/1`
Return `true` when the user context represents a websocket-originated session.
- `is_crawler/1`
Return `true` when the user context represents a crawler session.
- `is_anonymous_user/1`
Return `true` when the user context represents an anonymous user.
The last three callbacks are used by the session process to decide whether a
disconnected session should get the shortened timeout described above, without
inspecting the runtime-specific `user_context()` directly.
## Packet size protection
`mqtt_sessions` distinguishes between incoming and outgoing MQTT packet limits.
The incoming limit is controlled with the application env key
`max_incoming_packet_size`.
The outgoing limit is controlled with the application env key
`max_outgoing_packet_size`.
Incoming packets:
- incoming `CONNECT` packets larger than the incoming limit are refused
- incoming packets on established sessions are disconnected with
`packet_too_large`
- successful MQTT v5 `CONNACK` packets advertise the configured incoming
`maximum_packet_size` to the client
Outgoing packets:
- are unlimited by default from the server side
- can be capped with `max_outgoing_packet_size`
- always honor the client-advertised MQTT v5 `maximum_packet_size` from
`CONNECT`, if present
Defaults:
- `max_incoming_packet_size`: 20 MB
- `max_outgoing_packet_size`: unlimited
## Retained memory protection
Retained messages can be bounded with the application env key
`max_retained_memory`, expressed in bytes.
When configured, `mqtt_sessions`:
- removes expired retained messages first
- then evicts the oldest retained messages until the ETS memory used by the
retained-topic and retained-message tables is back within the configured limit
If `max_retained_memory` is not configured, then retained-message storage is not
explicitly configured, `mqtt_sessions` defaults it to 500 MB.
## Incoming publish rate limiting
`mqtt_sessions` can rate-limit incoming client `PUBLISH` packets per session
using:
- `max_incoming_messages_rate`
- `max_incoming_messages_burst`
`max_incoming_messages_rate` is the sustained number of incoming publish
messages per second. The period for this limit is `1` second.
`max_incoming_messages_burst` is the extra burst capacity on top of the regular
rate. This is not a separate time window; it is additional token-bucket
capacity that can be spent immediately and then refills at the regular per-
second rate.
The limiter uses a token-bucket model per session. This means a client can send
messages at the configured steady rate, while also consuming a limited burst
budget for short spikes.
When the bucket is empty, `mqtt_sessions` does not block inside the session
process. This is important because blocking there would let unrelated Erlang
messages pile up in the mailbox. Instead it fails fast to keep memory use
bounded:
- QoS 0 incoming publishes are dropped
- QoS 1 and QoS 2 incoming publishes are rejected with MQTT reason code
`Quota Exceeded`
Defaults:
- `max_incoming_messages_rate`: `1000`
- `max_incoming_messages_burst`: `5000`
### TODO
Current status of the list below:
- partially implemented:
`Connect rate`,
`Sessions with biggest queues`,
`Sessions with largest number of packets`
- not implemented:
`Number of connected sessions`,
`Number of packets sent / received`,
`Memory consumption of retained tables`
1. Add instrumentation
- Number of connected sessions
- Number of packets sent / received
- Connect rate
- Memory consumption of retained tables
- Sessions with biggest queues
- Sessions with largest number of packets
