diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 003cee2eb7..64b9e4e58e 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -614,6 +614,13 @@ actors: # + #
  • For MQTT Gateway (connected sub-devices): the same QoS rules as MQTT transport apply + # (only a QoS level 1 PUBACK triggers a delivery timeout). However, the gateway's own transport + # session is never closed, so the gateway connection and its other sub-devices stay unaffected. + # Only the sub-device session is deregistered and the RPC is reverted to the queued state. It is + # redelivered automatically once the sub-device session is re-registered, which happens on the next + # gateway message for that device (connect, telemetry, or attribute publish): the RPC subscription + # is reapplied and the queued RPC is resent. Until then the RPC remains queued.
    • #
  • For CoAP transport: #
    • diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java index 75d529d4b1..bb1037f481 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java @@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransp import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration; import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; import org.thingsboard.server.common.data.rpc.Rpc; +import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.msg.session.FeatureType; import org.thingsboard.server.gen.transport.TransportApiProtos; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; @@ -331,6 +332,62 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getMessageArrivedQoS()); } + protected void validateGatewayPersistentRpcDelivered(String deviceName) throws Exception { + GatewayRpcSession session = connectGatewayAndSubscribeForRpc(deviceName, false); + + long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); + String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"expirationTime\":" + expirationTime + "}"; + String response = doPostAsync("/api/rpc/twoway/" + session.device().getId().getId().toString(), rpcRequest, String.class, status().isOk()); + String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); + + session.callback().getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // Confirm the downlink arrived at QoS1 — otherwise no PUBACK is produced and the test could + // pass for the wrong reason if a regression downgraded the gateway RPC to QoS0. + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), session.callback().getMessageArrivedQoS()); + + awaitRpcStatus(rpcId, RpcStatus.DELIVERED, 200, 100); + + session.client().disconnect(); + } + + /** + * Connects a gateway, provisions the sub-device via CONNECT, and subscribes to the gateway RPC topic + * at QoS1. Two-step subscribe (client.subscribeAndWait + awaitForDeviceActorToReceiveSubscription, 1) + * rather than the 5-arg subscribeAndWait, which waits for subscription count+1 — the gateway CONNECT + * already registered one RPC subscription, so it would hang. + */ + protected GatewayRpcSession connectGatewayAndSubscribeForRpc(String deviceName, boolean manualAcks) throws Exception { + MqttTestClient client = new MqttTestClient(); + if (manualAcks) { + client.enableManualAcks(); + } + client.connectAndWait(gatewayAccessToken); + + String connectPayload = "{\"device\": \"" + deviceName + "\", \"type\": \"" + TransportPayloadType.JSON.name() + "\"}"; + client.publish(GATEWAY_CONNECT_TOPIC, connectPayload.getBytes()); + + Device device = doExecuteWithRetriesAndInterval(() -> getDeviceByName(deviceName), 20, 100); + assertNotNull(device); + + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_RPC_TOPIC); + client.setCallback(callback); + client.subscribeAndWait(GATEWAY_RPC_TOPIC, MqttQoS.AT_LEAST_ONCE); + awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.RPC, 1); + return new GatewayRpcSession(client, device, callback); + } + + protected Rpc awaitRpcStatus(String rpcId, RpcStatus expectedStatus, int retries, int intervalMs) throws Exception { + Rpc rpc = doExecuteWithRetriesAndInterval(() -> { + Rpc current = doGet("/api/rpc/persistent/" + rpcId, Rpc.class); + return expectedStatus.equals(current.getStatus()) ? current : null; + }, retries, intervalMs); + assertNotNull(rpc); + assertEquals(expectedStatus, rpc.getStatus()); + return rpc; + } + + protected record GatewayRpcSession(MqttTestClient client, Device device, MqttTestCallback callback) {} + protected void validateProtoTwoWayRpcGatewayResponse(String deviceName, MqttTestClient client, byte[] connectPayloadBytes) throws Exception { client.publish(GATEWAY_CONNECT_TOPIC, connectPayloadBytes); @@ -353,7 +410,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getMessageArrivedQoS()); } - private Device getDeviceByName(String deviceName) throws Exception { + protected Device getDeviceByName(String deviceName) throws Exception { return doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java new file mode 100644 index 0000000000..9c314339e5 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java @@ -0,0 +1,75 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.mqtt.mqttv3.rpc; + +import io.netty.handler.codec.mqtt.MqttQoS; +import lombok.extern.slf4j.Slf4j; +import org.junit.Before; +import org.junit.Test; +import org.springframework.test.context.TestPropertySource; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.TransportPayloadType; +import org.thingsboard.server.common.data.rpc.RpcStatus; +import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; + +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +@Slf4j +@DaoSqlTest +@TestPropertySource(properties = { + "actors.rpc.close_session_on_rpc_delivery_timeout=true", + "transport.mqtt.timeout=100", +}) +public class MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest extends AbstractMqttServerSideRpcIntegrationTest { + + @Before + public void beforeTest() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .deviceName("RPC timeout test device") + .gatewayName("RPC timeout test gateway") + .transportPayloadType(TransportPayloadType.JSON) + .build(); + processBeforeTest(configProperties); + } + + @Test + public void testGatewayPersistentRpcTimeoutWhenNoPuback() throws Exception { + // manualAcks = true → the gateway client never PUBACKs the QoS1 downlink. + GatewayRpcSession session = connectGatewayAndSubscribeForRpc("Gateway Device Persistent RPC Timeout Json", true); + + long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); + String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"retries\":0,\"expirationTime\":" + expirationTime + "}"; + String response = doPostAsync("/api/rpc/twoway/" + session.device().getId().getId().toString(), rpcRequest, String.class, status().isOk()); + String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); + + session.callback().getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // Confirm the downlink arrived at QoS1 — a QoS0 downlink would produce no PUBACK path at all, + // so the timeout would fire for the wrong reason and mask a regression. + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), session.callback().getMessageArrivedQoS()); + + // No PUBACK is ever sent. The gateway awaiting-ack scheduler emits TIMEOUT after + // transport.mqtt.timeout (100 ms); the RPC status reverts to QUEUED (re-queued via the + // close_session_on_rpc_delivery_timeout path). + awaitRpcStatus(rpcId, RpcStatus.QUEUED, 50, 100); + + session.client().disconnect(); + } + +} diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java index 10ef549c87..842904baef 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java @@ -81,6 +81,11 @@ public class MqttServerSideRpcJsonIntegrationTest extends AbstractMqttServerSide processJsonTwoWayRpcTestGateway("Gateway Device TwoWay RPC Json"); } + @Test + public void testGatewayServerMqttPersistentRpcDeliveredOnPuback() throws Exception { + validateGatewayPersistentRpcDelivered("Gateway Device Persistent RPC Json"); + } + protected void processJsonOneWayRpcTestGateway(String deviceName) throws Exception { MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index f753579292..7b321dc5f1 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -88,6 +88,7 @@ import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler; import org.thingsboard.server.transport.mqtt.session.MqttTopicMatcher; import org.thingsboard.server.transport.mqtt.session.SparkplugDeviceSessionContext; import org.thingsboard.server.transport.mqtt.session.SparkplugNodeSessionHandler; +import org.thingsboard.server.transport.mqtt.util.MqttRpcStatusUtil; import org.thingsboard.server.transport.mqtt.util.ReturnCodeResolver; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugRpcRequestHeader; @@ -385,6 +386,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId); if (rpcRequest != null) { transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); + } else if (gatewaySessionHandler != null) { + gatewaySessionHandler.onPubAck(msgId); } break; default: @@ -1521,9 +1524,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } public void sendToDeviceRpcRequest(MqttMessage payload, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) { - int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); + MqttPublishMessage publishMsg = (MqttPublishMessage) payload; + int msgId = publishMsg.variableHeader().packetId(); int requestId = rpcRequest.getRequestId(); - if (isAckExpected(payload)) { + boolean requireDeliveryTracking = MqttRpcStatusUtil.requireDeliveryTracking(rpcRequest); + if (requireDeliveryTracking && isAckExpected(publishMsg)) { rpcAwaitingAck.put(msgId, rpcRequest); context.getScheduler().schedule(() -> { TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId); @@ -1533,7 +1538,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } }, Math.max(0, Math.min(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS); } - var cf = publish(payload, deviceSessionCtx); + var cf = publish(publishMsg, deviceSessionCtx); cf.addListener(result -> { Throwable throwable = result.cause(); if (throwable != null) { @@ -1542,9 +1547,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement ThingsboardErrorCode.INVALID_ARGUMENTS, " Failed send To Device Rpc Request: " + rpcRequest.getMethodName()); return; } - if (!isAckExpected(payload)) { - log.trace("[{}][{}][{}] Going to send to device actor RPC request DELIVERED status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); - transportService.process(sessionInfo, rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + if (!isAckExpected(publishMsg)) { + if (requireDeliveryTracking) { + log.trace("[{}][{}][{}] Going to send to device actor RPC request DELIVERED status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); + transportService.process(sessionInfo, rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + } } else if (rpcRequest.getPersisted()) { log.trace("[{}][{}][{}] Going to send to device actor RPC request SENT status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); transportService.process(sessionInfo, rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index a1162f1990..5b0f8c15be 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -17,6 +17,7 @@ package org.thingsboard.server.transport.mqtt.session; import io.netty.channel.ChannelFuture; import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttPublishMessage; import lombok.Getter; import lombok.ToString; import lombok.extern.slf4j.Slf4j; @@ -29,6 +30,7 @@ import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; +import org.thingsboard.server.transport.mqtt.util.MqttRpcStatusUtil; import java.util.UUID; import java.util.concurrent.ConcurrentMap; @@ -106,19 +108,23 @@ public abstract class AbstractGatewayDeviceSessionContext { - ChannelFuture channelFuture = parent.writeAndFlush(payload); - if (request.getPersisted()) { - channelFuture.addListener(result -> { - if (result.cause() == null) { - if (!isAckExpected(payload)) { + MqttPublishMessage publishMsg = (MqttPublishMessage) payload; + boolean requireDeliveryTracking = MqttRpcStatusUtil.requireDeliveryTracking(request); + if (requireDeliveryTracking && isAckExpected(publishMsg)) { + parent.registerRpcAwaitingAck(publishMsg.variableHeader().packetId(), getSessionInfo(), request); + } + ChannelFuture channelFuture = parent.writeAndFlush(publishMsg); + channelFuture.addListener(result -> { + if (result.cause() == null) { + if (!isAckExpected(publishMsg)) { + if (requireDeliveryTracking) { transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); - } else if (request.getPersisted()) { - transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); - } + } else if (request.getPersisted()) { + transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); } - }); - } + } + }); } ); } catch (Exception e) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 8ea6480142..0a15d5d255 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -50,6 +50,7 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.data.util.TbPair; import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.common.msg.tools.TbRateLimitsException; @@ -80,6 +81,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; @@ -123,6 +125,10 @@ public abstract class AbstractGatewaySessionHandler rpcAwaitingAck = new ConcurrentHashMap<>(); + + private record RpcAwaitingAck(SessionInfoProto deviceSessionInfo, TransportProtos.ToDeviceRpcRequestMsg rpc) {} + @Getter @Setter private boolean overwriteDevicesActivity = false; @@ -201,6 +207,10 @@ public abstract class AbstractGatewaySessionHandler { Futures.addCallback(future, new FutureCallback() { @@ -252,6 +262,28 @@ public abstract class AbstractGatewaySessionHandler { + RpcAwaitingAck pending = rpcAwaitingAck.remove(msgId); + if (pending != null) { + log.trace("[{}] Gateway RPC [{}] PUBACK timeout, sending TIMEOUT status", sessionId, rpc.getRequestId()); + transportService.process(pending.deviceSessionInfo(), pending.rpc(), RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY); + } + }, delay, TimeUnit.MILLISECONDS); + } + + public void onPubAck(int msgId) { + RpcAwaitingAck pending = rpcAwaitingAck.remove(msgId); + if (pending == null) { + return; + } + log.trace("[{}] Gateway RPC [{}] PUBACK received, sending DELIVERED status", sessionId, pending.rpc().getRequestId()); + transportService.process(pending.deviceSessionInfo(), pending.rpc(), RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); + } + protected boolean isJsonPayloadType() { return deviceSessionCtx.isJsonPayloadType(); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/MqttRpcStatusUtil.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/MqttRpcStatusUtil.java new file mode 100644 index 0000000000..7aeb468d86 --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/MqttRpcStatusUtil.java @@ -0,0 +1,35 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.mqtt.util; + +import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg; + +public final class MqttRpcStatusUtil { + + private MqttRpcStatusUtil() { + } + + /** + * Whether the transport should track and emit a delivery status (DELIVERED / TIMEOUT) for the given RPC. + * Non-persistent one-way RPCs self-complete on send in the device actor and are never added to its pending + * map, so a delivery status for them lands on nothing and only produces a benign + * "RPC has already been removed from pending map" warning. Every other RPC is tracked. + */ + public static boolean requireDeliveryTracking(ToDeviceRpcRequestMsg rpc) { + return !(rpc.getOneway() && !rpc.getPersisted()); + } + +}