# Sparkplug B and OPC UA PubSub JSON: the plant floor under contract

> Sparkplug B nodes, host applications and OPC UA PubSub JSON publishers behind the gateway: roles, node table, failover, settings.

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

> 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 node or a publisher: [developer guide](https://flexway.io/en/docs/developer-mqtt-devices/#sparkplug-b-and-opc-ua-pubsub-json).

On the [MQTT device listener](https://flexway.io/en/docs/operations-mqtt-devices/), two industrial standards come under the same contract:
**Sparkplug B 3.0** - plant edge nodes, their devices, host applications (SCADA, MES) and their `STATE` - and **OPC UA
PubSub** in JSON over MQTT. The gateway judges each packet, relays it, then **reads** what crossed to keep the nodes'
state; it never speaks Sparkplug itself (no rebirth requested, no `STATE`, no death emitted). Available since `1.7.0`.

> **Conformance** - tested end to end against Mosquitto, EMQX (one node) and HiveMQ CE. The Sparkplug conformance kit
> (TCK 3.0.0) and the Eclipse Tahu nodes have not been run yet: no certified conformance is promised.

## Sparkplug B: the example

Edge nodes publish births, data and deaths; a host application of the same tenant commands them and announces its
`STATE`. Each tenant lives in **its** group of the `spBv1.0` namespace, each node under **its** identity.

<!-- example: examples/sparkplug/flexway.yaml -->

```yaml
# Sparkplug B 3.0 behind the gateway (1.7.0): edge nodes publish their births, data and deaths, a host
# application (SCADA, MES) commands them and announces its STATE - each tenant inside its own group of
# the namespace spBv1.0, every node's id bound to its identity. The gateway reads births,
# deaths and sequences into each replica's node table (GET /sparkplug/nodes, the console's Devices
# screen) and never speaks Sparkplug itself.
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
auth:
  api_keys:
    store: "file:./api-keys.yaml"                # a node's key id is the edge node id its topics carry
listeners:
  mqtt:
    address: ":1883"
    mesh_only: true                              # with TLS and device certificates, bind edge_node to cert:cn
    advertised_address: "mqtt.plant.example.com:1883"
observability:
  sparkplug:
    max_nodes: 10000                             # per replica; full of online nodes, a new one is counted, not kept
    offline_retention: 24h                       # an offline node stays listed this long
```

```yaml
# examples/sparkplug/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Plant Sparkplug B, version: "1.0.0" }
channels:
  edgeNode:
    # The edge node's own topics: its births, deaths and data, its devices', its commands. The + is the
    # message type, admitted by the operation's role; the group is the tenant, the node its identity.
    address: "spBv1.0/{group}/+/{edge_node}/#"
    parameters:
      group: { x-flexway-bind: tenant }
      edge_node: { x-flexway-bind: subject }       # the key's id; "cert:cn" on a TLS listener
  group:
    # The host application's view of its group.
    address: "spBv1.0/{group}/#"
    parameters:
      group: { x-flexway-bind: tenant }
  hostState:
    # STATE has no group: the host id is bound to the tenant, so two tenants never share one.
    address: "spBv1.0/STATE/{host_id}"
    parameters:
      host_id: { x-flexway-bind: tenant }
operations:
  nodePublishes:
    action: send                                 # NBIRTH, NDEATH, DBIRTH, DDEATH, NDATA, DDATA; the will: NDEATH only
    channel: { $ref: "#/channels/edgeNode" }
    x-flexway-upstream: plant
    x-flexway-validation: full                   # every payload under the embedded sparkplug_b.proto: 0x99 otherwise
    x-flexway-device: { qos_max: 1 }
    x-flexway-sparkplug: edge_node
    security: [ { type: userPassword } ]
  nodeCommands:
    action: receive                              # NCMD, DCMD addressed to the node
    channel: { $ref: "#/channels/edgeNode" }
    x-flexway-upstream: plant
    x-flexway-device: true
    x-flexway-sparkplug: edge_node
    security: [ { type: userPassword } ]
  nodeWatchesHost:
    action: receive                              # the primary host's STATE
    channel: { $ref: "#/channels/hostState" }
    x-flexway-upstream: plant
    x-flexway-device: true
    x-flexway-sparkplug: state
    security: [ { type: userPassword } ]
  hostCommands:
    action: send                                 # NCMD, DCMD only: a host never publishes a birth
    channel: { $ref: "#/channels/group" }
    x-flexway-upstream: plant
    x-flexway-device: { qos_max: 1 }
    x-flexway-sparkplug: host
    security: [ { type: userPassword } ]
  hostReads:
    action: receive
    channel: { $ref: "#/channels/group" }
    x-flexway-upstream: plant
    x-flexway-device: true
    x-flexway-sparkplug: host
    security: [ { type: userPassword } ]
  hostState:
    action: send                                 # {"online": true, "timestamp": ...}, its will the offline one
    channel: { $ref: "#/channels/hostState" }
    x-flexway-upstream: plant
    x-flexway-device: { qos_max: 1 }
    x-flexway-sparkplug: state
    security: [ { type: userPassword } ]
```

## Roles of a Sparkplug operation

A device operation (`x-flexway-device`) takes a **role** with `x-flexway-sparkplug`. Its channel has **exactly** the
role's shape, the group bound to the tenant - any other shape, and any channel in `spBv1.0/` without a role, is refused
at load: two tenants never meet in the namespace.

| Role | Channel | `send` | `receive` | Will admitted |
| --- | --- | --- | --- | --- |
| `edge_node` | `spBv1.0/{group}/+/{node}/#` - the `+` is the message type, the node bound to the identity (`subject`, `claim:<name>`, `cert:<field>`) | `NBIRTH`, `NDEATH`, `DBIRTH`, `DDEATH`, `NDATA`, `DDATA` | Its `NCMD`, `DCMD` | Its `NDEATH` only |
| `host` | `spBv1.0/{group}/#` | `NCMD`, `DCMD` only - never a birth | The whole group | None |
| `state` | `spBv1.0/STATE/{host_id}` - `host_id` bound to the **tenant** | Its `STATE` `{"online": ..., "timestamp": ...}` | Its tenant's `STATE` | Its `STATE` only |

- **One host application per tenant** through the gateway: `host_id` is the tenant's name (`spBv1.0/STATE/acme`). The
  `STATE` topic has no group; bound to anything else, two tenants would share one `STATE`.
- **Commands are mutations**: with a token, every alternative of the `security` of a `host` operation in `send` names
  its scopes (refused at load otherwise); a key or a certificate is bound to the operation.
- A tenant named `STATE` never gets a Sparkplug operation: its host channel would cover every tenant's `STATE`.
- **Payload**: a Sparkplug operation without a message gets the B payload under the embedded `sparkplug_b.proto`, or
  `STATE`'s JSON schema. With `x-flexway-validation: full`, every payload is checked against it (`0x99` otherwise);
  without it, a malformed payload is relayed and changes nothing in the table.

On a TLS listener, bind the node to its certificate:

```yaml
channels:
  edgeNode:
    address: "spBv1.0/{group}/+/{edge_node}/#"
    parameters:
      group: { x-flexway-bind: tenant }
      edge_node: { x-flexway-bind: "cert:cn" }   # the device certificate's common name
operations:
  nodePublishes:
    action: send
    channel: { $ref: "#/channels/edgeNode" }
    x-flexway-upstream: plant
    x-flexway-device: { qos_max: 1 }
    x-flexway-sparkplug: edge_node
    security: [ { type: X509 } ]
```

## The node table

Each replica keeps the table of the edge nodes and host applications that cross it, read **after** the relay: `seq` and
the `bdSeq` metric only, never delaying nor refusing an admitted packet.

| State | When |
| --- | --- |
| `online` | Its `NBIRTH` crossed this replica, its connection is open here |
| `offline` | Its `NDEATH`, or the end of its connection with its will |
| `moved` | This replica handed its connection over to another (failover): a trace, **never** offline |
| `unknown` | Only commands to it were seen here |

- **Gaps**: a `seq` that is not the next one, or a birth's `bdSeq` that differs from its will's, are counted - never
  blocking.
- **Commands**: `NCMD`/`DCMD` admitted and refused counted per node (a refusal only on a node of its own tenant).
  Commands published through the HTTP door `x-flexway-publish` are not tracked.
- **Host applications**: their `STATE` is read from their publication, their will and the retained `STATE` delivered
  to the tenant's devices.
- **Bounds**: when full, the table forgets an `unknown` node first, then the oldest `offline` or `moved` one; full of
  online nodes, a new one is not kept and is counted (`dropped`). It is lost at a restart.

```bash
curl -s 'localhost:9090/sparkplug/nodes?tenant=acme'
# {"replica":"flexway-7c9d-x2","capacity":10000,"offline_retention_seconds":86400,"dropped":0,
#  "nodes":[{"tenant":"acme","edge_node":"press-line-1","state":"online","since":"2026-10-23T08:12:04Z",
#    "bd_seq":4,"will_bd_seq":4,"seq_gaps":0,"bd_seq_gaps":0,"protocol_version":"5","auth":"api_key",
#    "devices":[{"device":"press-07","state":"online"}],"devices_truncated":false,
#    "commands":{"admitted":3,"refused":0}}],
#  "hosts":[{"tenant":"acme","host_id":"acme","state":"online","since":"2026-10-23T08:11:58Z"}]}
```

`GET /sparkplug/nodes` lives on the admin port (`404` without an MQTT listener); `bd_seq` and `will_bd_seq` are absent
when nothing was read. The console reads every replica's table and merges them
([Console: the Devices screen](https://flexway.io/en/docs/operations-console-devices/)).

## Host applications connected directly

`NCMD`/`DCMD` commands are governed - scopes, audit, counts - only when the host application goes **through the
gateway**. A host application connected directly to the broker escapes all of it: bound it with the broker's ACLs
(publish only on `spBv1.0/<its group>/NCMD/#` and `DCMD/#`, `STATE` on its own topic only).

## Lost replica, shutdown and failover

| Case | What the broker and the console see |
| --- | --- |
| Draining shutdown | The node comes back through another replica with the same client id: the broker resumes its session, no `NDEATH`; the old replica keeps a `moved` trace, the console shows the node online and **rehomed** |
| Lost replica | Its connections drop: the broker publishes a **true** `NDEATH` - it follows a gateway failure, not the node's. No `moved` trace: the node is reborn (`NBIRTH`, `bdSeq` + 1) online on the replica that receives it |
| Alert | `sparkplug-node-offline` waits 2 min (`for`) on the merged table: a node back in the meantime fires nothing |

Will on takeover, `handover_timeout` and `will_delay_on_handover`: [Failover](https://flexway.io/en/docs/operations-mqtt-devices/#failover-shutdown-and-full-replica).

## Brokers

| Broker | Tested | Good to know |
| --- | --- | --- |
| Mosquitto 2 | Yes | - |
| EMQX 6 | Yes, **one node** | BSL 1.1 licence since 5.9: one node free, a cluster under a commercial licence |
| HiveMQ CE | Yes | Carries the Sparkplug conformance kit (a HiveMQ extension) |

Sparkplug's "aware" server (`$sparkplug/certificates/...`) stays with the broker: the gateway does not do it.

## OPC UA PubSub JSON

No extension and no OPC UA stack: a publisher sending its JSON NetworkMessages (Part 14, `ua-data`) over MQTT is a
device under an **ordinary contract**. Part 14's topic, `<prefix>/json/data/<PublisherId>/<WriterGroup>`: the tenant in
the prefix, the PublisherId bound to the identity; the message validated by its JSON Schema (`0x99` otherwise). The UADP
binary encoding is not served.

<!-- example: examples/opcua-pubsub/flexway.yaml -->

```yaml
# OPC UA PubSub over MQTT with the JSON message mapping (OPC UA Part 14, 1.7.0): publishers of a plant
# send their NetworkMessages through the gateway, each under its tenant and its own publisher id, each
# message checked against the contract's schema; subscribers read their tenant's data. No OPC UA stack
# in the gateway: a NetworkMessage is a JSON payload under a JSON Schema, like any other.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  plant:
    protocol: mqtt
    brokers: ["mqtts://broker.plant.svc.cluster.local:8883"]
    timeout: 5s
auth:
  api_keys:
    store: "file:./api-keys.yaml"                # the key's id is the publisher id the topic carries
listeners:
  mqtt:
    address: ":1883"
    mesh_only: true
    advertised_address: "mqtt.plant.example.com:1883"
```

```yaml
# examples/opcua-pubsub/asyncapi.yaml
asyncapi: 3.0.0
info: { title: OPC UA PubSub JSON, version: "1.0.0" }
channels:
  data:
    # Part 14's topic: <prefix>/json/data/<PublisherId>/<WriterGroup>, the prefix carrying the tenant.
    address: "opcua/{site}/json/data/{publisher}/line"
    parameters:
      site: { x-flexway-bind: tenant }
      publisher: { x-flexway-bind: subject }
    messages:
      networkMessage: { $ref: "#/components/messages/networkMessage" }
  allData:
    address: "opcua/{site}/json/data/#"
    parameters:
      site: { x-flexway-bind: tenant }
    messages:
      networkMessage: { $ref: "#/components/messages/networkMessage" }
operations:
  publishData:
    action: send
    channel: { $ref: "#/channels/data" }
    x-flexway-upstream: plant
    x-flexway-validation: full
    x-flexway-device: { qos_max: 1 }
    security: [ { type: userPassword } ]
  readData:
    action: receive
    channel: { $ref: "#/channels/allData" }
    x-flexway-upstream: plant
    x-flexway-device: { qos_max: 1 }
    security: [ { type: userPassword } ]
components:
  messages:
    networkMessage:
      contentType: application/json
      payload:
        # A NetworkMessage of the JSON mapping (OPC UA Part 14, section 7.2.5.3): ua-data, its
        # DataSetMessages and their fields.
        type: object
        required: [MessageId, MessageType, Messages]
        properties:
          MessageId: { type: string, maxLength: 64 }
          MessageType: { const: ua-data }
          PublisherId: { type: string, maxLength: 128 }
          WriterGroupName: { type: string }
          Messages:
            type: array
            minItems: 1
            maxItems: 64
            items:
              type: object
              required: [Payload]
              properties:
                DataSetWriterId: { type: integer, minimum: 0, maximum: 65535 }
                SequenceNumber: { type: integer, minimum: 0 }
                Timestamp: { type: string, format: date-time }
                MessageType: { enum: [ua-keyframe, ua-deltaframe, ua-event, ua-keepalive] }
                Payload: { type: object, maxProperties: 256 }
```

## Settings and defaults

The listener's (`listeners.mqtt.*`, `failover.*`, `observability.mqtt.closures`, chart `mqtt.enabled`):
[Settings and defaults](https://flexway.io/en/docs/operations-mqtt-devices/#settings-and-defaults). The console's (five built-in rules,
`devices.readerBounds`): [Console: the Devices screen](https://flexway.io/en/docs/operations-console-devices/). Sparkplug's own, reloaded live:

| Key (`observability.sparkplug.`) | Default | Why |
| --- | --- | --- |
| `max_nodes` | 10,000 per replica (1 to 50,000: a node is a connection) | Edge nodes and host applications kept; the table's memory bound |
| `max_nodes_per_tenant` | 1,000, or `max_nodes` when lower (1 to `max_nodes`) | One tenant's nodes: a tenant never fills the others' table; at its bound its own unknown, then oldest offline nodes are forgotten, otherwise the new one is counted (`dropped`) |
| `offline_retention` | `24h` (1 min to 720 h) | How long an `offline`, `moved` or `unknown` node stays listed |
| `max_devices_per_node` | 1,000 (1 to 10,000) | Devices (`DBIRTH`, `DDEATH`) kept per node; past it, the list is said truncated |

A group, node, device or host id longer than **255 bytes** is no Sparkplug topic for the gateway: the publication is
refused (`0x87`) and nothing is kept. A refused command never creates an entry in the table. A tenant name starting
with `_` is the gateway's own (`_anonymous`, `_other` in the metrics): a key, a certificate client or a claim carrying
one is refused.

```yaml
observability:
  sparkplug:
    max_nodes: 20000
    max_nodes_per_tenant: 2000
    offline_retention: 72h          # a weekend without a node still shows it
    max_devices_per_node: 1000
```

## Observing

| Signal | What it says |
| --- | --- |
| `flexway.sparkplug.nodes{tenant, state}` | Nodes of the table, `online` and `offline`; a `moved` node is neither |
| `flexway.sparkplug.sequence.gaps{tenant, kind}` | `seq` and `bdseq` gaps |
| `flexway.mqtt.packets{type, decision, cause}` | Packets refused by the role (`not_authorized`) or the schema (`payload_invalid`) |
| `GET /sparkplug/nodes` | The names - never in a metric |

## Troubleshooting

| What the node or the operator sees | Why | Where to look |
| --- | --- | --- |
| Refused at load: `a x-flexway-sparkplug edge_node operation's channel is spBv1.0/{group}/+/{edge_node}/#` | The channel does not have the role's shape, or the group is not bound to the tenant | The table of roles |
| Refused at load: `writes ... in the Sparkplug namespace spBv1.0/` | A channel in `spBv1.0/` without `x-flexway-sparkplug` | Mark the operation with its role |
| Refused at load: `admits a token without scopes` | A `host` operation in `send` admits a token without scopes | Name the commands' scopes |
| A node's `CONNACK 0x87` | Its will is not its `NDEATH` (or, for `state`, its `STATE`) | The will configured in the node |
| `PUBACK 0x87` | A type the role does not publish (a birth by the host, a command by the node), or another's group or node | `POST /simulate/mqtt` with the topic |
| `PUBACK 0x99` | Payload outside `sparkplug_b.proto` or the OPC UA schema (`x-flexway-validation: full`) | The payload sent |
| Node missing from the table | Commands sent through HTTP (not tracked), table full (`dropped`), replica restarted | `GET /sparkplug/nodes` of every replica |
| Node `offline` after a replica failure | The broker's `NDEATH` is true; the node is reborn elsewhere as soon as it reconnects | The console: online on another replica |