# MQTT devices: the listener in proxy in front of your broker

> MQTT 5 and 3.1.1 devices behind the gateway, in front of your broker: identity, topics bound to the tenant, backpressure, failover.

Concept - Since version 1.7.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-mqtt-devices/

> Operations guide: configure, deploy and observe the gateway. What the feature does for the product: [product guide](https://flexway.io/en/docs/product-iot-devices/); connecting a device: [developer guide](https://flexway.io/en/docs/developer-mqtt-devices/).

Sensors, controllers and plant gateways connect to the gateway in **MQTT 5 or 3.1.1**, over TLS or WebSocket, instead
of your broker. The gateway is a **proxy**, never a broker: each device has its own connection to your broker
(Mosquitto, EMQX, HiveMQ...), opened in the gateway's name; each packet is judged against the AsyncAPI contract - who
the device is, which topics its tenant and identity open, its rates and quota - then relayed as it is. Sessions,
retained messages, wills and QoS 2 stay at your broker. Available since `1.7.0`.

> **On this listener**: [Sparkplug B and OPC UA PubSub JSON](https://flexway.io/en/docs/operations-sparkplug/); the console's Devices screen:
> [Console: the Devices screen](https://flexway.io/en/docs/operations-console-devices/). The defaults of `max_connections`,
> `max_connections_per_ip` and `max_connection_rate_per_ip` are provisional, pending performance measurements: write
> them down if you depend on them.

<!-- example: examples/mqtt-devices/flexway.yaml -->

```yaml
# Devices behind the gateway (1.7.0): sensors and controllers connect in MQTT 5 or 3.1.1 and reach the
# plant's broker through the gateway - one broker connection per device, each packet judged against the
# AsyncAPI contract, each tenant inside its own topics. Here clear text inside a mesh, devices by API key
# or token; the TLS listener and device certificates are in the guide.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  plant:
    protocol: mqtt
    brokers: ["mqtts://broker.plant.svc.cluster.local:8883"]
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/broker/password" }
    timeout: 5s                                  # a connection to the broker, per device
auth:
  issuers:
    - name: devices
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: plant-devices
      tenant_claim: site                         # every topic a device reaches is its tenant's
  api_keys:
    store: "file:./api-keys.yaml"                # the key's id as the user name, the key as the password
listeners:
  mqtt:
    address: ":1883"
    mesh_only: true                              # clear text only where the mesh encrypts between pods
    websocket: { address: ":8884" }              # MQTT over WebSocket, path /mqtt
    advertised_address: "mqtt.plant.example.com:1883"   # the load balancer: where a drained device goes
    device:
      messages_per_second: 50                    # past it the device is held, never refused
    failover: { handover_timeout: 30s }
```

```yaml
# examples/mqtt-devices/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Plant devices, version: "1.0.0" }
channels:
  telemetry:
    address: "plant/{site}/{device}/telemetry"
    parameters:
      site: { x-flexway-bind: tenant }           # the verified tenant, never read from the topic
      device: { x-flexway-bind: subject }        # the key's id, the token's subject
    messages:
      reading:
        payload: { type: object, required: [v], properties: { v: { type: number } } }
  commands:
    address: "plant/{site}/{device}/cmd"
    parameters:
      site: { x-flexway-bind: tenant }
      device: { x-flexway-bind: subject }
    messages: { command: { payload: { type: object } } }
  fleet:
    address: "plant/{site}/fleet/#"                    # a wildcard after the bound level: the tenant's fleet
    parameters:
      site: { x-flexway-bind: tenant }
    messages: { event: { payload: { type: object } } }
operations:
  publishTelemetry:
    action: send                                 # the device publishes
    channel: { $ref: "#/channels/telemetry" }
    x-flexway-upstream: plant
    x-flexway-validation: full                   # a reading against its schema: 0x99 otherwise
    x-flexway-device: { qos_max: 1 }
    security:
      - type: userPassword
      - type: oauth2
        flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "telemetry:write": publish } } }
        scopes: ["telemetry:write"]
  readCommands:
    action: receive                              # the device subscribes
    channel: { $ref: "#/channels/commands" }
    x-flexway-upstream: plant
    x-flexway-device: true
    security: [ { type: userPassword } ]
  readFleet:
    action: receive
    channel: { $ref: "#/channels/fleet" }
    x-flexway-upstream: plant
    x-flexway-device: { shared: true, qos_max: 1 }   # $share/<group>/plant/<site>/fleet/#, the group in the tenant's name
    security: [ { type: userPassword } ]
```

## Declaring what a device may do

An AsyncAPI operation marked `x-flexway-device` opens its channel to devices, in **its** direction: `send`, the device
publishes; `receive`, it subscribes. Nothing else passes.

| Key of `x-flexway-device` | Default | Effect |
| --- | --- | --- |
| `qos_max` | 2 | Highest QoS; a `PUBLISH` past it is refused `0x9B`, a subscription granted at it |
| `messages_per_second`, `bytes_per_second` | those of `listeners.mqtt.device` | The rates of a `PUBLISH` on this channel (`send`) |
| `shared` | `false` | Admits `$share/<group>/<filter>` on a `receive` channel; the group is written `<tenant>~<group>` at the broker |

Every parameter of the channel is **bound** to a verified fact (`x-flexway-bind`) and one of them to the **tenant** -
refused at load otherwise: `tenant`, `subject` (the key's id, the token's subject, the `auth.mtls` client's id),
`claim:<name>`, or a field of the device certificate, `cert:<field>`: `cn`, `serial`, `fingerprint` (the certificate's
SHA-256, hexadecimal), or with its index `ou[i]`, `o[i]`, `san.dns[i]`, `san.uri[i]`, `san.ip[i]`, `san.email[i]`. A
value that is empty or holds `/`, `+`, `#`, `$`, `{`, `}` or a NUL keeps the operation closed: never written into a
topic, and a value written is never read again as a parameter. A wildcard (`+`, `#`) is written in a `receive` channel
only, after the bound levels: a device's filter must be **included** in it.

All the device channels write the **tenant at the same level**, after the same literal prefix (`plant/{site}/...`
everywhere) - refused at load otherwise: a channel with the tenant at level 1 and another with it at level 2 would let
the tenant named like a literal level of one read the other's topics.

## Identifying devices

The listener reads, in order: the **certificate** verified by `tls.client_ca_file` and declared in `auth.mtls`, else the
`CONNECT`'s **password** - an **API key** whose id is the user name, else a **token** (`exp` required; when it is
reached, `DISCONNECT 0xA0`). An alternative of the operation's `security` names what it admits: `X509`, `userPassword`
(or `plain`, `apiKey` `in: password`), `oauth2`/`openIdConnect`.

Device certificates (TLS required):

```yaml
listeners:
  mqtt:
    address: ":8883"
    tls:
      cert_file: /etc/flexway-mqtt-tls/tls.crt
      key_file: /etc/flexway-mqtt-tls/tls.key
      client_ca_file: /etc/flexway-mqtt-tls/ca.crt   # verifies device certificates
    websocket: { address: ":8884", path: /mqtt }      # MQTT over WebSocket, same TLS
    advertised_address: "mqtt.plant.example.com:8883"
auth:
  mtls:
    clients:
      - { id: press-07, subject: "CN=press-07", tenant: acme, operations: [publishCertified] }
```

```yaml
channels:
  certified:
    address: "plant/{site}/{cn}/certified/{dns}"
    parameters:
      site: { x-flexway-bind: tenant }
      cn:   { x-flexway-bind: "cert:cn" }           # the device certificate's common name
      dns:  { x-flexway-bind: "cert:san.dns[0]" }   # its first DNS name: a multi-valued field names its index
operations:
  publishCertified:
    action: send
    channel: { $ref: "#/channels/certified" }
    x-flexway-upstream: plant
    x-flexway-device: true
    security: [ { type: X509 } ]
```

Tokens of an OIDC provider, the tenant in a claim:

```yaml
auth:
  issuers:
    - name: devices
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: plant-devices
      tenant_claim: site
```

At your broker, the device has the client id **`<tenant>/<mechanism>:<identity>`** (`acme/apikey:press-07`), never the
one it chose: a device of one tenant never evicts another's. A token's carries a digest of its issuer,
`acme/oidc:<8 hex>:<subject>`: two issuers handing out the same subject never share a session. A second connection of
the same identity takes the session over (`0x8E` to the first); `device.sessions: per_client_id` gives a session per
device client id under a shared identity, at most `device.max_sessions_per_identity` at once (`0x97` past it).

> **`subject` and several issuers** - `x-flexway-bind: subject` binds the subject alone: two issuers of the same tenant
> that hand out the same subject reach the same topics. Give each tenant one issuer, or bind a claim only one issuer
> sets (`claim:<name>`). An `issuer` binding is proposed for a later lot, not built.

**Keep-alive.** An MQTT 5 device is told `max_keepalive` (Server Keep Alive) when it asks more, or none. MQTT 3.1.1
cannot be told one: a device asking more than `max_keepalive` is refused `0x05`, and one asking none (`0`) too, unless
`allow_zero_keepalive: true` - a silent device would never be seen gone. Every connection ends after
`max_session_lifetime` (24 h) whatever its credential, and reconnects with a credential checked again.

**Revocation.** At each reload, every open connection is checked again within half a second against the new
configuration - a device that only listens too: a key, client or operation withdrawn ends it with `DISCONNECT 0x87`,
and its broker connection with `DISCONNECT 0x00`, so the broker discards its will (a revoked device is not a lost one).
Each delivery from the broker is judged on the configuration in force.

## Settings and defaults

Every default can be changed; the bounds apply to the next packet or `CONNECT`, except the address, TLS, `mesh_only`,
the WebSocket and `max_connections`, read at start.

| Key (`listeners.mqtt.`) | Default | Why |
| --- | --- | --- |
| `address`, `tls`, `mesh_only` | no listener unless declared; TLS or `mesh_only` | A device's token or key never crosses the network in clear outside a mesh |
| `broker_mesh_only` | `false` | A broker in clear (`mqtt://`, `ws://`) is refused at load outside `mesh_only`: set it when a mesh encrypts the hop to the broker |
| `websocket.path` | `/mqtt` | The path of MQTT-over-WebSocket clients (browsers, plant gateways) |
| `max_packet_bytes` | 256 KiB | A packet is read whole before it is judged: the memory bound per packet |
| `connect_timeout` | `5s` | A slow `CONNECT` or a lingering TLS handshake never holds a place |
| `max_connections` | 50,000 per replica | Provisional: to be confirmed by bench M1; past it `0x97` |
| `max_unauthenticated_connections` | 1,024 | A burst without `CONNECT` never starves authenticated devices |
| `max_connections_per_ip`, `max_connection_rate_per_ip` | 64; 20 per second | Provisional (bench M1); set `0` behind a load balancer or proxy that hides addresses (the chart keeps them: `externalTrafficPolicy: Local`) - otherwise every device shares one address and one bound |
| `max_connections_per_tenant`, `max_connection_rate_per_tenant` | 10,000; 100 per second | One tenant never takes the replica from the others; a reconnection storm is spread |
| `max_keepalive`, `max_session_expiry`, `topic_alias_maximum` | 1,200 s; 1 h; 10 (1,000 at most) | The bounds of the market's brokers (AWS, Event Grid, HiveMQ); an alias is recorded only once its topic is granted |
| `allow_zero_keepalive` | `false` | A 3.1.1 device asking no keep-alive is refused: it would never be seen gone |
| `max_session_lifetime` | 24 h (1 min to 30 days) | A connection never outlives a revocation the reload would miss |
| `device.messages_per_second`, `bytes_per_second`, `burst` | 100; 512 KiB; 2 | AWS IoT Core's; backpressure past them, never a refusal |
| `device.receive_maximum`, `max_subscriptions`, `max_filters_per_subscribe` | 100; 50; 10 | Bound what a device holds in flight and subscribes |
| `device.connect_rate_per_identity` | 1 per second | A device in a loop never evicts itself again and again (`0x9F`); keyed on the verified identity |
| `device.max_refusals_per_minute` | 60 | Each refusal spends the device's message rate, its audit is sampled (one entry a second per cause); past it `DISCONNECT 0x96` |
| `device.max_user_properties_bytes`, `max_correlation_data_bytes` | 8 KiB; 8 KiB | What a packet carries beside its payload |
| `device.sessions` | `single` | One session per identity: a second device evicts the first (`0x8E`) |
| `device.max_sessions_per_identity` | 10 (or `max_connections` if lower) | With `per_client_id`, one identity never takes the replica (`0x97`) |
| `tenant.messages_per_second`, `bytes_per_second` | 5,000; 20 MiB | The ceiling of a whole site; subscription plans come in `1.9.0` |
| `failover.mode`, `reason` | `redirect`; `use_another_server` | A shutdown redirects rather than cuts; `0x9C` is temporary |
| `failover.server_reference` | `advertised_address` | Always reachable, even by a client that ignores the reference |
| `failover.redirect_above`, `handover_timeout` | 95 %; 30 s | Redirects before saturation; lets a common client with backoff come back |
| `failover.will_delay_on_handover` | `false` | On: an MQTT 5 will without delay keeps through a takeover, a lost device's will goes up to `handover_timeout` later |
| `server.shutdown.mqtt_drain_spread` | 30 s (half of `drain` at most) | No reconnection storm on the other replicas |
| `observability.mqtt.closures` | `true` | Count normal closures apart from refusals |

## Rates, backpressure and quota

Past a rate of the device or of its tenant, the gateway **stops reading** the device's socket for as long as needed -
never a refusal, never a closure that would fire its will - and **keeps its keep-alive** toward the broker meanwhile. The
tenant's day or month quota (a `quota` policy, as for HTTP) is a refusal: `PUBACK 0x97`, QoS 0 dropped and counted,
3.1.1 closed.

`device.max_subscriptions` counts the filters a device subscribes on this connection. A device that comes back with
`Session Present=1` keeps at the broker the subscriptions of its previous connection, which the gateway does not see:
bound them at the broker too (its per-client subscription limit), the gateway's bound being per connection.

```yaml
listeners:
  mqtt:
    device:
      messages_per_second: 100      # past it: held, never refused
      bytes_per_second: 524288
      burst: 2                      # seconds of each rate a burst may take
    tenant:
      messages_per_second: 5000
      bytes_per_second: 20971520
      quota: plant-monthly          # a policy of type quota: 0x97 once the month is spent
```

## Failover: shutdown and full replica

When a replica shuts down, each MQTT 5 device receives `DISCONNECT 0x9C` and the **Server Reference** (the load
balancer's address by default); a 3.1.1 device is closed and reconnects through the load balancer. Its broker connection
is **held** for `handover_timeout`: back through another replica, the broker takes its session over and **no will** is
published; not back, the connection is closed with `DISCONNECT 0x04` and the will goes - the device is really gone. Past
`redirect_above`, a new device is redirected at its `CONNECT`. A replica lost abruptly cuts its connections: the broker
publishes the wills, which are true.

```yaml
server:
  shutdown:
    drain: 45s                      # covers the spread and the handover
    mqtt_drain_spread: 30s
listeners:
  mqtt:
    advertised_address: "mqtt.plant.example.com:8883"
    failover:
      mode: redirect                # off: 0x8B and the will at once
      reason: use_another_server    # server_moved (0x9D) for a planned migration
      redirect_above: 95            # % of max_connections: new devices go elsewhere first
      handover_timeout: 30s
      will_delay_on_handover: false # true: a will without delay gets handover_timeout (R11)
```

The pod's grace period follows `server.shutdown.drain`: cover the spread and the handover.

> **Will at takeover** - Mosquitto, EMQX and HiveMQ CE publish the will of a session **taken over** when its Will
> Delay Interval is 0: that is MQTT 5's rule, not the gateway's. The takeover keeps the session (subscriptions,
> messages in flight); to keep the will too, an MQTT 5 device declares a Will Delay Interval longer than its reconnection
> and a Session Expiry Interval that covers it (20 s and 40 s for instance). A 3.1.1 device has no delay: its will goes
> at the takeover. `flexway.mqtt.failover{outcome="taken_over"}` counts these takeovers apart.
>
> For devices you cannot change, `failover.will_delay_on_handover: true` (off by default) has the gateway write toward
> the broker a Will Delay Interval of `handover_timeout`, and a session expiry that covers it, when the will declares
> none: the will survives the takeover. **The cost**: the will of a device really lost goes up to `handover_timeout`
> later. No effect on an MQTT 3.1.1 device - its broker connection is 3.1.1, which has no will delay.

## The chart

The container's ports, the `Service` and its `NetworkPolicy` follow `listeners.mqtt` and are rendered only with
`mqtt.enabled` (off by default); the cert-manager `Certificate` with `listeners.mqtt.tls`. The certificate is mounted at
`/etc/flexway-mqtt-tls` (`tls.crt`, `tls.key`, `ca.crt`). With `mesh_only`, the `Service` stays `ClusterIP` and
`mqtt.networkPolicy.from` is required, `mqtt.enabled` or not: the process listens in clear either way. A `LoadBalancer`
(or `NodePort`) `Service` gets `externalTrafficPolicy: Local` (`mqtt.service.externalTrafficPolicy`): the devices'
addresses reach the gateway, and `max_connections_per_ip` bounds a device - `Cluster` hides them, then set
`max_connections_per_ip: 0`. The WebSocket listener counts its sockets from their acceptance, before TLS and the
upgrade.

```yaml
mqtt:
  enabled: true                     # the Service; off by default
  service: { type: LoadBalancer }   # externalTrafficPolicy: Local by default
  tls:
    dnsNames: [mqtt.plant.example.com]
    issuerRef: { name: devices-ca, kind: ClusterIssuer }
  networkPolicy:
    from: [{ ipBlock: { cidr: 10.20.0.0/16 } }]
```

An address per replica (a `Service` per pod, for a Server Reference per replica) is not rendered: the load balancer is
enough, and many clients ignore the reference.

## Observing and simulating

| Signal | What it says |
| --- | --- |
| `flexway.mqtt.connections{listener, tenant, protocol_version, auth}` | Devices connected: the occupation of `max_connections` |
| `flexway.mqtt.packets{type, decision, cause, tenant}` | Packets admitted, refused, dropped, and why |
| `flexway.mqtt.connections.refused{bound}`, `flexway.mqtt.connections.closed{cause}` | Refusals by bound; normal closures apart |
| `flexway.mqtt.publish.duration{tenant, qos}` | From the device's `PUBLISH` to the broker's acknowledgement |
| `flexway.mqtt.throttled`, `flexway.mqtt.throttled.devices` | Backpressure holds; devices held now |
| `flexway.mqtt.failover{reason, outcome}` | Redirections, takeovers, expired handovers |
| `flexway.mqtt.upstream.connections{outcome}` | Connections to the broker, circuit breaker included |

A span per sampled packet; each refusal in the decision audit (`audit.decisions`). `/config` publishes `listeners.mqtt`
and `devices` (the operations, their channels and bindings). `POST /simulate/mqtt` judges a described packet without
sending anything to the broker:

```bash
curl -s localhost:9090/simulate/mqtt -d '{
  "packet": "subscribe", "topic": "$share/workers/plant/acme/fleet/#", "qos": 2,
  "identity": { "kind": "api_key", "id": "press-07" } }'
# {"verdict":"served","operation":"readFleet","upstream":"plant",
#  "written":"$share/acme~workers/plant/acme/fleet/#","qos":1,"client_id":"acme/apikey:press-07"}
```

## Troubleshooting

| What the device sees | Why | Where to look |
| --- | --- | --- |
| `CONNACK 0x86` (3.1.1 `0x04`) | Key or token refused, a user name that is not the key's id, a token without `exp` | The log line `mqtt connection refused` says why; `flexway.mqtt.packets{type="connect",cause="authentication"}` |
| `CONNACK 0x87` (3.1.1 `0x05`) | Identity without a tenant, a certificate not declared in `auth.mtls`, no operation admits it, a will outside its topics | Check the key's or client's `operations`, the will channel's `x-flexway-bind` |
| `CONNACK 0x8C` | The client asks for enhanced authentication (`AUTH`) | Not served: key, token or certificate |
| `CONNACK 0x88` (3.1.1 `0x03`) | Broker unreachable or circuit breaker open | `flexway.mqtt.upstream.connections{outcome}`, the upstream's `brokers` |
| `CONNACK 0x97`, `0x9F` | `max_connections`, a tenant bound or `max_sessions_per_identity`; connection rate of the tenant or the identity | `flexway.mqtt.connections.refused{bound}` |
| 3.1.1 return code `0x05` at once | A keep-alive of `0` or past `max_keepalive` | Set the device's keep-alive, or `allow_zero_keepalive` |
| `CONNACK 0x9C` + Server Reference | Replica past `redirect_above` | Expected: the client goes to `server_reference` |
| `PUBACK 0x87` | Topic outside its channels (`binding`, `not_authorized`), a `$` topic, a Response Topic outside its subscriptions | `POST /simulate/mqtt` with the topic and the identity |
| `PUBACK 0x99` | Payload refused by the schema (`x-flexway-validation: full`), or declared UTF-8 (Payload Format Indicator 1) and not | The contract's message |
| `PUBACK 0x97` | The tenant's day or month quota is spent - never a rate | `listeners.mqtt.tenant.quota` |
| `PUBACK`/`PUBREC 0x9B` | QoS past the operation's `qos_max` | `x-flexway-device.qos_max` |
| `SUBACK 0x87` (3.1.1 `0x80`) | Filter outside a channel it reads (`wildcard`, `binding`), `$share/` on a channel not `shared` | `POST /simulate/mqtt` with `packet: subscribe` |
| `DISCONNECT 0x93`, `0x94`, `0x95` | Receive Maximum exceeded, unknown alias or past the maximum, packet too large | The `device.*` bounds and `max_packet_bytes` |
| `DISCONNECT 0xA0` | The credential ended (token, key, certificate) or `max_session_lifetime` | The client reconnects with a fresh credential |
| `DISCONNECT 0x87` | A reload withdrew its key, client or operation | `/config`, the key store |
| `DISCONNECT 0x96` | More than `device.max_refusals_per_minute` packets refused | The refusals in `flexway.mqtt.packets{decision="refused"}` and the sampled audit |
| `DISCONNECT 0x8E` | Another connection of the same identity took the session over | `device.sessions: per_client_id` for several devices under one identity |