MQTT reference connector

The first concrete connector instantiating the framework’s contracts. This cluster :satisfies: Connector framework (FEAT_0030).

Feature: MQTT reference connector FEAT_0036
status: open
satisfies: FEAT_0030
is satisfied by: REQ_0250, REQ_0251, REQ_0252, REQ_0253, REQ_0254, REQ_0255, REQ_0256, REQ_0257, REQ_0258, REQ_0259, REQ_0260, REQ_0261, REQ_0980, REQ_0981, REQ_0982, REQ_0983, REQ_0984, REQ_0985, REQ_0986, REQ_0987, REQ_0988
is refined by: ADR_0010, ADR_0128, ADR_0129, ADR_0130, ADR_0131

Motivation. The first concrete connector instantiating the framework’s contracts — a rumqttc-backed MQTT 3.1.1 plugin and gateway that exercises every seam (codec, routing, health, reconnect, bounded bridges) so the framework core is proven against a real external protocol.

Scope. Bidirectional pub/sub over MQTT 3.1.1; QoS 0+1; retained publish; wildcard subscriptions with gateway-local fan-out; optional TLS; username/password auth. Reconnection is delegated to rumqttc’s EventLoop and surfaced as ConnectorHealth (rumqttc EventLoop owns MQTT... (ADR_0128)), including a terminal-Down policy. Connections use a clean session with SUBSCRIBE replay on reconnect (Clean session with SUBSCRIB... (ADR_0130)). Inbound demultiplexing runs a gateway-local topic matcher over deduplicated, reference-counted broker subscriptions (Wildcard inbound demux via ... (ADR_0129)). JsonCodec is the default codec (JSON default codec; MsgPack... (ADR_0131)).

Non-goals (deferred to follow-on specs). QoS 2; MQTT 5.0 features (user properties, shared subscriptions, response topic); Last-Will-and-Testament; persistent (clean_session=false) sessions; client-certificate authentication. MsgPackCodec is a prerequisite change to taktora-connector-codec (JSON default codec; MsgPack... (ADR_0131)) that must land and publish independently before this connector cites it — it is not part of this connector’s landing.
Requirement: MqttConnector implements Connector REQ_0250
status: implemented
github: 168
satisfies: FEAT_0036
is implemented by: BB_0004, BB_0020
is verified by: TEST_0957
links outgoing: BB_0004, TEST_0957
links incoming: ADR_0131

The connector crate shall expose MqttConnector<C: PayloadCodec> that implements the Connector trait with type Routing = MqttRouting.
Requirement: MqttRouting carries topic, qos, retained REQ_0251
status: implemented
github: 167
satisfies: FEAT_0036
is implemented by: BB_0004, BB_0020
is verified by: TEST_0958
links outgoing: BB_0004, TEST_0958

The MqttRouting struct shall carry the MQTT topic name, the QoS level, and a retained-message flag. It shall implement the Routing marker trait.
Requirement: QoS 0 and 1 supported REQ_0252
status: implemented
github: 168
satisfies: FEAT_0036
is verified by: TEST_0961
links outgoing: BB_0004, TEST_0961

The connector shall support MQTT QoS levels AtMostOnce (0) and AtLeastOnce (1). QoS 2 is deferred to a follow-on spec.
Requirement: Retained-message publish supported REQ_0253
status: implemented
github: 168
satisfies: FEAT_0036
is verified by: TEST_0961
links outgoing: BB_0004, TEST_0961

When MqttRouting::retained is true, the connector shall publish the envelope payload as a retained MQTT message.
Requirement: Wildcard subscriptions supported REQ_0254
status: implemented
github: 169
satisfies: FEAT_0036
is refined by: ARCH_0011
is verified by: TEST_0102, TEST_0959, TEST_0960, TEST_0966
links outgoing: BB_0004, TEST_0959, TEST_0960, TEST_0966
links incoming: RISK_0005, ADR_0129

The connector shall accept inbound subscriptions whose topic includes the MQTT wildcards + (single-level) and # (multi-level), and shall demultiplex received messages to the matching ChannelReader instance(s).
Requirement: Username/password authentication REQ_0255
status: implemented
github: 170
satisfies: FEAT_0036
is verified by: TEST_0255
links outgoing: BB_0004, TEST_0255

The connector shall accept username and password credentials in MqttConnectorOptions and present them on the MQTT CONNECT packet.
Requirement: TLS is optional via cargo feature REQ_0256
status: implemented
github: 170
satisfies: FEAT_0036
is verified by: TEST_0256
links outgoing: BB_0004, TEST_0256

The connector shall provide TLS support via rustls behind a default-off tls cargo feature. Client-certificate authentication is deferred to a follow-on spec.
Requirement: MQTT 3.1.1 baseline REQ_0257
status: implemented
github: 170
satisfies: FEAT_0036
is verified by: TEST_0257
links outgoing: BB_0004, TEST_0257

The connector shall target MQTT protocol version 3.1.1. MQTT 5.0 features (user properties, shared subscriptions, response topic) are deferred to a follow-on spec.
Requirement: Tokio sidecar inside the gateway crate REQ_0258
status: implemented
github: 168
satisfies: FEAT_0036
is implemented by: BB_0004, BB_0021
is verified by: TEST_0962
links outgoing: BB_0004, TEST_0962

The MQTT gateway shall host rumqttc::EventLoop on a tokio runtime contained inside taktora-connector-mqtt. Tokio shall not leak into taktora-executor’s WaitSet thread.
Requirement: Bridge channels are bounded REQ_0259
status: implemented
github: 167
satisfies: FEAT_0036
is implemented by: BB_0021, BB_0022
is verified by: TEST_0160, TEST_0963
links outgoing: BB_0004, TEST_0963

The outbound (taktora-executor → tokio) and inbound (tokio → taktora-executor) bridges shall be bounded channels with configurable capacity in MqttConnectorOptions.
Requirement: Outbound bridge saturation surfaces as BackPressure REQ_0260
status: implemented
github: 168
satisfies: FEAT_0036
is refined by: ARCH_0033
is implemented by: BB_0021, BB_0022
is verified by: TEST_0964
links outgoing: BB_0004, TEST_0964
links incoming: ADR_0128

When the outbound bridge channel is full, ChannelWriter::send shall return ConnectorError::BackPressure and the connector shall report ConnectorHealth::Degraded.
Requirement: Inbound bridge saturation drops frames and signals Degraded REQ_0261
status: implemented
github: 169
satisfies: FEAT_0036
is refined by: ARCH_0033
is implemented by: BB_0021, BB_0022
is verified by: TEST_0965
links outgoing: BB_0004, TEST_0965

When the inbound bridge channel is full, the gateway shall (1) increment the per-channel inbound-drop counter exposed via InboundOutcome::Dropped { count } on the bridge’s try_send return, (2) drop the offending message for that delivery, and (3) emit a ConnectorHealth::Degraded { reason: "dropped N inbound frames" } health transition when the cumulative inbound-drop count crosses the connector’s configured inbound_drop_threshold (default 1). The Degraded transition is emitted at most once until the connector recovers to Up via the underlying stack’s recovery path; the cumulative drop count itself is observable through every subsequent InboundOutcome::Dropped return.

Reconnection and health

The connector delegates reconnection to rumqttc and observes liveness as health rather than repairing it in the connector (rumqttc EventLoop owns MQTT... (ADR_0128)).

Requirement: Connection state maps to ConnectorHealth REQ_0980
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0968
links outgoing: BB_0004, TEST_0968

The gateway shall map the MQTT connection state onto ConnectorHealth: a pending or backing-off connection attempt shall report Connecting and a successful CONNACK shall report Up.
Requirement: Reconnect backoff is configurable REQ_0981
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0969
links outgoing: BB_0004, TEST_0969

The connector shall expose rumqttc’s reconnect backoff parameters through MqttConnectorOptions rather than a bespoke reconnect loop.
Requirement: Auth-rejected CONNACK transitions to Down REQ_0982
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0970
links outgoing: BB_0004, TEST_0970

When the broker returns a CONNACK with an authentication or authorization failure return code, the connector shall transition to ConnectorHealth::Down without further reconnect attempts.
Requirement: Reconnect-attempt ceiling transitions to Down REQ_0983
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0971
links outgoing: BB_0004, TEST_0971

The connector shall transition to ConnectorHealth::Down when the number of consecutive failed reconnect attempts exceeds a configurable ceiling in MqttConnectorOptions.

Session model

The connector uses a clean session and replays its subscriptions on every reconnect (Clean session with SUBSCRIB... (ADR_0130)).

Requirement: Clean session on CONNECT REQ_0984
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0972
links outgoing: BB_0004, TEST_0972

The connector shall connect with the MQTT clean-session flag set to true. The flag shall be configurable via MqttConnectorOptions.
Requirement: SUBSCRIBE replay on reconnect REQ_0985
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0973
links outgoing: BB_0004, TEST_0973

On each reconnect CONNACK the gateway shall replay every active subscription from its subscription table, since the clean session retains no broker-side subscription state.

Wildcard demux mechanism

These requirements refine Wildcard subscriptions supp... (REQ_0254) with the demultiplexing mechanism (Wildcard inbound demux via ... (ADR_0129)).

Requirement: Broker subscriptions are deduplicated and reference-counted REQ_0986
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0967
links outgoing: BB_0004, TEST_0967

The gateway shall register each distinct topic filter with the broker at most once and reference-count the channels using it. It shall send UNSUBSCRIBE for a filter when the last channel referencing it is removed — by explicit channel removal or at connector teardown. Per-ChannelReader-drop teardown is out of scope: the Connector::create_reader contract returns an un-hooked reader handle, so subscription lifetime is bounded by the connector, matching the framework’s other pub/sub connector (see Wildcard inbound demux via ... (ADR_0129)).
Requirement: Inbound PUBLISH is matched locally and fanned out REQ_0987
status: implemented
github: 169
satisfies: FEAT_0036
is verified by: TEST_0960, TEST_0966
links outgoing: BB_0004, TEST_0960, TEST_0966

The gateway shall match each inbound PUBLISH topic locally against all registered channel filters and deliver the message to every matching ChannelReader instance.

Codec default

Requirement: JsonCodec is the default codec REQ_0988
status: implemented
github: 167
satisfies: FEAT_0036
is verified by: TEST_0974
links outgoing: BB_0004, TEST_0974

The MQTT connector’s examples and integration tests shall use JsonCodec as the default codec (JSON default codec; MsgPack... (ADR_0131)).