diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index d8710eaa2c..21defb9f6a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -193,17 +193,18 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso log.debug("[{}][{}] Received RPC request to process ...", deviceId, rpcId); ToDeviceRpcRequestMsg rpcRequest = createToDeviceRpcRequestMsg(request); - long timeout = request.getExpirationTime() - System.currentTimeMillis(); + long createdTime = System.currentTimeMillis(); + long timeout = request.getExpirationTime() - createdTime; boolean persisted = request.isPersisted(); if (timeout <= 0) { log.debug("[{}][{}] Ignoring message due to exp time reached, {}", deviceId, rpcId, request.getExpirationTime()); if (persisted) { - createRpc(request, RpcStatus.EXPIRED); + createRpc(request, RpcStatus.EXPIRED, createdTime); } return; } else if (persisted) { - createRpc(request, RpcStatus.QUEUED); + createRpc(request, RpcStatus.QUEUED, createdTime); } boolean sent = false; @@ -243,7 +244,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso log.debug("[{}] RPC command response sent [{}][{}]!", deviceId, rpcId, requestId); systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, null)); } else { - registerPendingRpcRequest(context, msg, sent, rpcRequest, timeout); + registerPendingRpcRequest(context, msg, sent, rpcRequest, timeout, createdTime); } String rpcSent = sent ? "sent!" : "NOT sent!"; log.debug("[{}][{}][{}] RPC request is {}", deviceId, rpcId, requestId, rpcSent); @@ -257,16 +258,21 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso }; } - private void createRpc(ToDeviceRpcRequest request, RpcStatus status) { + private void createRpc(ToDeviceRpcRequest request, RpcStatus status, long createdTime) { + systemContext.getTbRpcService().create(tenantId, buildRpc(request, status, null, createdTime)); + } + + private Rpc buildRpc(ToDeviceRpcRequest request, RpcStatus status, JsonNode response, long createdTime) { Rpc rpc = new Rpc(new RpcId(request.getId())); - rpc.setCreatedTime(System.currentTimeMillis()); + rpc.setCreatedTime(createdTime); rpc.setTenantId(tenantId); rpc.setDeviceId(deviceId); rpc.setExpirationTime(request.getExpirationTime()); rpc.setRequest(JacksonUtil.valueToTree(request)); + rpc.setResponse(response); rpc.setStatus(status); rpc.setAdditionalInfo(getAdditionalInfo(request)); - systemContext.getTbRpcService().save(tenantId, rpc); + return rpc; } private JsonNode getAdditionalInfo(ToDeviceRpcRequest request) { @@ -334,11 +340,11 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } } - private void registerPendingRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) { + private void registerPendingRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout, long createdTime) { int requestId = rpcRequest.getRequestId(); UUID rpcId = new UUID(rpcRequest.getRequestIdMSB(), rpcRequest.getRequestIdLSB()); log.debug("[{}][{}][{}] Registering pending RPC request...", deviceId, rpcId, requestId); - toDeviceRpcPendingMap.put(requestId, new ToDeviceRpcRequestMetadata(msg, sent)); + toDeviceRpcPendingMap.put(requestId, new ToDeviceRpcRequestMetadata(msg, sent, createdTime)); DeviceActorServerSideRpcTimeoutMsg timeoutMsg = new DeviceActorServerSideRpcTimeoutMsg(requestId, timeout); scheduleMsgWithDelay(context, timeoutMsg, timeoutMsg.getTimeout()); } @@ -351,7 +357,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso UUID rpcId = toDeviceRpcRequest.getId(); log.debug("[{}][{}][{}] RPC request timeout detected!", deviceId, rpcId, requestId); if (toDeviceRpcRequest.isPersisted()) { - systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), RpcStatus.EXPIRED, null); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.EXPIRED, null, requestMd.getCreatedTime())); } systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION)); @@ -662,7 +668,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } catch (IllegalArgumentException e) { response = JacksonUtil.newObjectNode().put("error", payload); } - systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, response); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRequestMsg, status, response, requestMd.getCreatedTime())); } } finally { if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE)) { @@ -726,7 +732,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } if (persisted) { - systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, response); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, status, response, md.getCreatedTime())); } if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE) && status.equals(RpcStatus.DELIVERED) && !oneWayRpc) { @@ -819,7 +825,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso var toDeviceRpcRequest = md.getMsg().getMsg(); if (toDeviceRpcRequest.isPersisted()) { var responseAwaitTimeout = JacksonUtil.newObjectNode().put("error", "There was a timeout awaiting for RPC response from device."); - systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), RpcStatus.FAILED, responseAwaitTimeout); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout, md.getCreatedTime())); } }, systemContext.getRpcResponseTimeout(), TimeUnit.MILLISECONDS); } @@ -1030,9 +1036,9 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso long timeout = rpc.getExpirationTime() - System.currentTimeMillis(); if (timeout <= 0) { rpc.setStatus(RpcStatus.EXPIRED); - systemContext.getTbRpcService().save(tenantId, rpc); + systemContext.getTbRpcService().update(tenantId, rpc); } else { - registerPendingRpcRequest(ctx, new ToDeviceRpcRequestActorMsg(systemContext.getServiceId(), msg), false, createToDeviceRpcRequestMsg(msg), timeout); + registerPendingRpcRequest(ctx, new ToDeviceRpcRequestActorMsg(systemContext.getServiceId(), msg), false, createToDeviceRpcRequestMsg(msg), timeout, rpc.getCreatedTime()); } }); if (pageData.hasNext()) { diff --git a/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java b/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java index 782fdd92ad..c502ce8baa 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java @@ -25,6 +25,9 @@ import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequestActorMsg; public class ToDeviceRpcRequestMetadata { private final ToDeviceRpcRequestActorMsg msg; private final boolean sent; + // Creation time of the persisted RPC row, captured once at create so post-persist rule-engine + // notifications on the update paths carry the original createdTime instead of the update moment. + private final long createdTime; private int retries; private boolean delivered; } diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java index 912e8242c6..c7c9a1e188 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java @@ -15,14 +15,17 @@ */ package org.thingsboard.server.service.rpc; -import com.fasterxml.jackson.databind.JsonNode; -import lombok.RequiredArgsConstructor; +import com.google.common.util.concurrent.ListenableFuture; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.common.util.HashPartitioner; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.id.DeviceId; -import org.thingsboard.server.common.data.id.RpcId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.page.PageData; @@ -34,31 +37,75 @@ import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.rpc.RpcService; import org.thingsboard.server.queue.util.TbCoreComponent; +import java.util.UUID; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + @TbCoreComponent @Service -@RequiredArgsConstructor @Slf4j public class TbRpcService { private final RpcService rpcService; private final TbClusterService tbClusterService; - public Rpc save(TenantId tenantId, Rpc rpc) { - Rpc saved = rpcService.save(rpc); - pushRpcMsgToRuleEngine(tenantId, saved); - return saved; + private final ExecutorService[] callbackExecutors; + + public TbRpcService(RpcService rpcService, TbClusterService tbClusterService, + @Value("${sql.rpc.callback_threads:3}") int callbackThreads) { + if (callbackThreads < 1) { + throw new IllegalArgumentException("sql.rpc.callback_threads must be >= 1, but was " + callbackThreads); + } + this.rpcService = rpcService; + this.tbClusterService = tbClusterService; + this.callbackExecutors = new ExecutorService[callbackThreads]; + for (int i = 0; i < callbackThreads; i++) { + callbackExecutors[i] = Executors.newSingleThreadExecutor( + ThingsBoardThreadFactory.forName("rpc-persist-callback-" + i)); + } } - public void save(TenantId tenantId, RpcId rpcId, RpcStatus newStatus, JsonNode response) { - Rpc foundRpc = rpcService.findById(tenantId, rpcId); - if (foundRpc != null) { - foundRpc.setStatus(newStatus); - if (response != null) { - foundRpc.setResponse(response); - } - Rpc saved = rpcService.save(foundRpc); - pushRpcMsgToRuleEngine(tenantId, saved); - } else { - log.warn("[{}] Failed to update RPC status because RPC was already deleted", rpcId); + @PreDestroy + private void destroy() { + for (ExecutorService executor : callbackExecutors) { + executor.shutdownNow(); + } + } + + public void create(TenantId tenantId, Rpc rpc) { + rpcService.save(rpc); + executorFor(rpc.getUuidId()).execute(() -> notifyRuleEngine(tenantId, rpc)); + } + + public void update(TenantId tenantId, Rpc rpc) { + persist(tenantId, rpc, rpcService.updateAsync(rpc)); + } + + private void persist(TenantId tenantId, Rpc rpc, ListenableFuture future) { + DonAsynchron.withCallback(future, + persisted -> { + if (Boolean.TRUE.equals(persisted)) { + notifyRuleEngine(tenantId, rpc); + } else { + log.debug("[{}][{}][{}] Skipping rule engine notification for status [{}] - RPC row no longer exists", + tenantId, rpc.getDeviceId(), rpc.getId(), rpc.getStatus()); + } + }, + t -> log.error("[{}][{}][{}] Failed to persist RPC with status [{}]", + tenantId, rpc.getDeviceId(), rpc.getId(), rpc.getStatus(), t), + executorFor(rpc.getUuidId())); + } + + private Executor executorFor(UUID rpcId) { + return callbackExecutors[HashPartitioner.resolvePartition(rpcId.hashCode(), callbackExecutors.length)]; + } + + private void notifyRuleEngine(TenantId tenantId, Rpc rpc) { + try { + pushRpcMsgToRuleEngine(tenantId, rpc); + } catch (Throwable t) { + log.error("[{}][{}][{}] Failed to push RPC with status [{}] to rule engine", + tenantId, rpc.getDeviceId(), rpc.getId(), rpc.getStatus(), t); } } @@ -72,10 +119,6 @@ public class TbRpcService { tbClusterService.pushMsgToRuleEngine(tenantId, rpc.getDeviceId(), msg, null); } - public Rpc findRpcById(TenantId tenantId, RpcId rpcId) { - return rpcService.findById(tenantId, rpcId); - } - public PageData findAllByDeviceIdAndStatus(TenantId tenantId, DeviceId deviceId, RpcStatus rpcStatus, PageLink pageLink) { return rpcService.findAllByDeviceIdAndStatus(tenantId, deviceId, rpcStatus, pageLink); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 9b0ca8d847..05b5a3e3cf 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -447,6 +447,12 @@ sql: stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}" # Interval in milliseconds for printing latest telemetry updates statistic batch_threads: "${SQL_TS_LATEST_BATCH_THREADS:3}" # batch thread count has to be a prime number like 3 or 5 to gain perfect hash distribution update_by_latest_ts: "${SQL_TS_UPDATE_BY_LATEST_TIMESTAMP:true}" # Update latest values only if the timestamp of the new record is greater or equals the timestamp of the previously saved latest value. The latest values are stored separately from historical values for fast lookup from DB. Insert of historical value happens in any case + rpc: + batch_size: "${SQL_RPC_BATCH_SIZE:1000}" # Batch size for persisting RPC status updates + batch_max_delay: "${SQL_RPC_BATCH_MAX_DELAY_MS:50}" # Max timeout for RPC entries queue polling, in milliseconds + stats_print_interval_ms: "${SQL_RPC_BATCH_STATS_PRINT_MS:10000}" # Interval in milliseconds for printing RPC persistence statistic + batch_threads: "${SQL_RPC_BATCH_THREADS:3}" # number of queue partitions for batched RPC persistence; writes for a given RPC always map to the same partition (by rpcId hash) so they persist in submission order. A prime value (e.g. 3 or 5) keeps the hash distribution even + callback_threads: "${SQL_RPC_CALLBACK_THREADS:3}" # striped threads for post-persist rule-engine notifications; each rpcId maps to a single stripe, preserving per-command notification order (e.g. RPC_QUEUED before RPC_DELIVERED). Independent of batch_threads above: that pool partitions DB writes, this one partitions notifications, and each stripes by rpcId on its own - so the two can be sized independently without affecting ordering. A prime value (e.g. 3 or 5) keeps the hash distribution even events: batch_size: "${SQL_EVENTS_BATCH_SIZE:10000}" # Batch size for persisting latest telemetry updates batch_max_delay: "${SQL_EVENTS_BATCH_MAX_DELAY_MS:100}" # Max timeout for latest telemetry entries queue polling. The value set in milliseconds diff --git a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java new file mode 100644 index 0000000000..bdb194e1de --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -0,0 +1,153 @@ +/** + * 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.service.rpc; + +import com.google.common.util.concurrent.Futures; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.RpcId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.msg.TbMsgType; +import org.thingsboard.server.common.data.rpc.Rpc; +import org.thingsboard.server.common.data.rpc.RpcStatus; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.dao.rpc.RpcService; + +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.after; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class TbRpcServiceTest { + + private final RpcService rpcService = mock(RpcService.class); + private final TbClusterService clusterService = mock(TbClusterService.class); + private TbRpcService tbRpcService; + + @Before + public void setUp() { + tbRpcService = new TbRpcService(rpcService, clusterService, 1); + } + + @Test + public void updatePersistsViaUpdateAsyncThenPushesToRuleEngine() { + Rpc rpc = newRpc(); + when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(true)); + + tbRpcService.update(rpc.getTenantId(), rpc); + + verify(rpcService).updateAsync(rpc); + verify(clusterService, timeout(5000)) + .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); + } + + @Test + public void createPersistsSynchronouslyThenPushesToRuleEngine() { + Rpc rpc = newRpc(); + when(rpcService.save(rpc)).thenReturn(rpc); + + tbRpcService.create(rpc.getTenantId(), rpc); + + // create must persist synchronously via save(...) + verify(rpcService).save(rpc); + verify(clusterService, timeout(5000)) + .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); + } + + @Test + public void updateDoesNotNotifyRuleEngineWhenRowMissing() { + Rpc rpc = newRpc(); + // updateAsync resolves false: the UPDATE matched no row (RPC was deleted), so the rule engine + // must not be notified for a status change that never persisted. + when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(false)); + + tbRpcService.update(rpc.getTenantId(), rpc); + + verify(rpcService).updateAsync(rpc); + verify(clusterService, after(500).never()) + .pushMsgToRuleEngine(any(TenantId.class), any(DeviceId.class), any(TbMsg.class), isNull()); + } + + @Test + public void sameRpcIdUpdateNotificationsRunInSubmissionOrderOnStripedExecutor() throws InterruptedException { + // A single rpcId always maps to one stripe; this checks that two update notifications for the + // SAME rpcId are serialized on that stripe (DELIVERED before SUCCESSFUL). + tbRpcService = new TbRpcService(rpcService, clusterService, 3); + + RpcId rpcId = new RpcId(UUID.randomUUID()); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + Rpc delivered = newRpc(rpcId, deviceId, RpcStatus.DELIVERED); + Rpc successful = newRpc(rpcId, deviceId, RpcStatus.SUCCESSFUL); + when(rpcService.updateAsync(delivered)).thenReturn(Futures.immediateFuture(true)); + when(rpcService.updateAsync(successful)).thenReturn(Futures.immediateFuture(true)); + + // Make the DELIVERED notification block while it runs (signal start, then sleep) - holding the + // stripe. If the two callbacks for this rpcId were NOT serialized on one stripe, the fast + // SUCCESSFUL one would overtake the sleeping DELIVERED one and be recorded first. + CountDownLatch deliveredStarted = new CountDownLatch(1); + doAnswer(invocation -> { + TbMsg msg = invocation.getArgument(2); + if (msg.getInternalType() == TbMsgType.RPC_DELIVERED) { + deliveredStarted.countDown(); + Thread.sleep(300); + } + return null; + }).when(clusterService).pushMsgToRuleEngine(eq(TenantId.SYS_TENANT_ID), eq(deviceId), any(TbMsg.class), isNull()); + + tbRpcService.update(delivered.getTenantId(), delivered); + // Don't submit SUCCESSFUL until DELIVERED is in flight (and now sleeping) on the stripe. + assertTrue(deliveredStarted.await(5, TimeUnit.SECONDS)); + tbRpcService.update(successful.getTenantId(), successful); + + ArgumentCaptor msgCaptor = ArgumentCaptor.forClass(TbMsg.class); + verify(clusterService, timeout(5000).times(2)) + .pushMsgToRuleEngine(eq(TenantId.SYS_TENANT_ID), eq(deviceId), msgCaptor.capture(), isNull()); + + List msgs = msgCaptor.getAllValues(); + assertEquals(TbMsgType.RPC_DELIVERED, msgs.get(0).getInternalType()); + assertEquals(TbMsgType.RPC_SUCCESSFUL, msgs.get(1).getInternalType()); + } + + private Rpc newRpc() { + return newRpc(new RpcId(UUID.randomUUID()), new DeviceId(UUID.randomUUID()), RpcStatus.QUEUED); + } + + private Rpc newRpc(RpcId rpcId, DeviceId deviceId, RpcStatus status) { + Rpc rpc = new Rpc(rpcId); + rpc.setTenantId(TenantId.SYS_TENANT_ID); + rpc.setDeviceId(deviceId); + rpc.setStatus(status); + rpc.setRequest(JacksonUtil.toJsonNode("{}")); + return rpc; + } +} diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java index 855063d657..19defc8eea 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java @@ -29,6 +29,8 @@ public interface RpcService extends EntityDaoService { Rpc save(Rpc rpc); + ListenableFuture updateAsync(Rpc rpc); + void deleteRpc(TenantId tenantId, RpcId id); void deleteAllRpcByTenantId(TenantId tenantId); diff --git a/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java b/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java new file mode 100644 index 0000000000..69efbd7ca4 --- /dev/null +++ b/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java @@ -0,0 +1,32 @@ +/** + * 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.common.util; + +/** + * Maps a hash code to a non-negative partition index in {@code [0, partitions)}. + */ +public final class HashPartitioner { + + private HashPartitioner() { + } + + public static int resolvePartition(int hashCode, int partitions) { + if (partitions <= 0) { + throw new IllegalArgumentException("partitions must be > 0, but was " + partitions); + } + return (hashCode & 0x7FFFFFFF) % partitions; + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java b/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java index b096c69d4b..44a265b26d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java @@ -54,6 +54,12 @@ public class BaseRpcService implements RpcService { return rpcDao.save(rpc.getTenantId(), rpc); } + @Override + public ListenableFuture updateAsync(Rpc rpc) { + log.trace("Executing updateAsync, [{}]", rpc); + return rpcDao.updateAsync(rpc); + } + @Override public void deleteRpc(TenantId tenantId, RpcId rpcId) { log.trace("Executing deleteRpc, tenantId [{}], rpcId [{}]", tenantId, rpcId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java index 37fe2950b4..c3999b2a11 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.dao.rpc; +import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; @@ -25,6 +26,8 @@ import org.thingsboard.server.dao.Dao; public interface RpcDao extends Dao { + ListenableFuture updateAsync(Rpc rpc); + PageData findAllByDeviceId(TenantId tenantId, DeviceId deviceId, PageLink pageLink); PageData findAllByDeviceIdAndStatus(TenantId tenantId, DeviceId deviceId, RpcStatus rpcStatus, PageLink pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java b/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java index 4ae8661d77..1dab15b473 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java @@ -18,6 +18,7 @@ package org.thingsboard.server.dao.sql; import com.google.common.util.concurrent.ListenableFuture; import lombok.Data; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.HashPartitioner; import org.thingsboard.server.common.stats.MessagesStats; import org.thingsboard.server.common.stats.StatsFactory; @@ -58,7 +59,7 @@ public class TbSqlBlockingQueueWrapper { } public ListenableFuture add(E element) { - int queueIndex = element != null ? (hashCodeFunction.apply(element) & 0x7FFFFFFF) % maxThreads : 0; + int queueIndex = element != null ? HashPartitioner.resolvePartition(hashCodeFunction.apply(element), maxThreads) : 0; return queues.get(queueIndex).add(element); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java index a929b6d1e4..fcdfa17088 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java @@ -15,8 +15,12 @@ */ package org.thingsboard.server.dao.sql.rpc; -import lombok.AllArgsConstructor; +import com.google.common.util.concurrent.ListenableFuture; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Transactional; @@ -27,22 +31,74 @@ import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rpc.Rpc; import org.thingsboard.server.common.data.rpc.RpcStatus; +import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.TenantEntityDao; import org.thingsboard.server.dao.model.sql.RpcEntity; import org.thingsboard.server.dao.rpc.RpcDao; import org.thingsboard.server.dao.sql.JpaAbstractDao; +import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent; +import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams; +import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper; import org.thingsboard.server.dao.util.SqlDao; +import java.util.Comparator; import java.util.UUID; +import java.util.function.Function; @Slf4j @Component -@AllArgsConstructor @SqlDao +@RequiredArgsConstructor public class JpaRpcDao extends JpaAbstractDao implements RpcDao, TenantEntityDao { private final RpcRepository rpcRepository; + private final RpcUpdateRepository rpcUpdateRepository; + private final ScheduledLogExecutorComponent logExecutor; + private final StatsFactory statsFactory; + + @Value("${sql.rpc.batch_size:1000}") + private int batchSize; + @Value("${sql.rpc.batch_max_delay:50}") + private long maxDelay; + @Value("${sql.rpc.stats_print_interval_ms:10000}") + private long statsPrintIntervalMs; + @Value("${sql.rpc.batch_threads:3}") + private int batchThreads; + @Value("${sql.batch_sort:true}") + private boolean batchSortEnabled; + + private TbSqlBlockingQueueWrapper queue; + + @PostConstruct + private void init() { + TbSqlBlockingQueueParams params = TbSqlBlockingQueueParams.builder() + .logName("RPC") + .batchSize(batchSize) + .maxDelay(maxDelay) + .statsPrintIntervalMs(statsPrintIntervalMs) + .statsNamePrefix("rpc") + .batchSortEnabled(batchSortEnabled) + .withResponse(true) + .build(); + Function hashcodeFunction = entity -> entity.getUuid().hashCode(); + queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory); + queue.init(logExecutor, entries -> rpcUpdateRepository.update(entries), + Comparator.comparing(RpcEntity::getUuid), + Function.identity()); + } + + @PreDestroy + private void destroy() { + if (queue != null) { + queue.destroy(); + } + } + + @Override + public ListenableFuture updateAsync(Rpc rpc) { + return queue.add(new RpcEntity(rpc)); + } @Override protected Class getEntityClass() { diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java new file mode 100644 index 0000000000..9403e551fb --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java @@ -0,0 +1,64 @@ +/** + * 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.dao.sql.rpc; + +import com.fasterxml.jackson.databind.JsonNode; +import org.springframework.jdbc.core.BatchPreparedStatementSetter; +import org.springframework.stereotype.Repository; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.dao.model.sql.RpcEntity; +import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository; + +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; + +@Repository +public class RpcUpdateRepository extends AbstractInsertRepository { + + private static final String UPDATE = + "UPDATE rpc SET status = ?, response = COALESCE(?, response) WHERE id = ?;"; + + List update(List updates) { + return transactionTemplate.execute(status -> { + int[] updateCounts = jdbcTemplate.batchUpdate(UPDATE, new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + RpcEntity rpc = updates.get(i); + ps.setString(1, rpc.getStatus().name()); + ps.setString(2, toJsonStr(rpc.getResponse())); + ps.setObject(3, rpc.getUuid()); + } + + @Override + public int getBatchSize() { + return updates.size(); + } + }); + + List persisted = new ArrayList<>(updateCounts.length); + for (int updateCount : updateCounts) { + persisted.add(updateCount > 0); + } + return persisted; + }); + } + + private String toJsonStr(JsonNode node) { + return node == null ? null : replaceNullChars(JacksonUtil.toString(node)); + } +} diff --git a/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java b/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java index 921339a92b..35dbfabfb3 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java @@ -15,16 +15,21 @@ */ package org.thingsboard.server.dao.sql.rpc; +import com.fasterxml.jackson.databind.JsonNode; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.RpcId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.rpc.Rpc; import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.dao.AbstractJpaDaoTest; +import org.thingsboard.server.dao.model.sql.RpcEntity; +import java.util.List; import java.util.UUID; +import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; @@ -33,6 +38,9 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { @Autowired JpaRpcDao rpcDao; + @Autowired + RpcUpdateRepository rpcUpdateRepository; + @Test public void deleteOutdated() { Rpc rpc = new Rpc(); @@ -57,4 +65,128 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { assertThat(rpcDao.deleteOutdatedRpcByTenantIdBatch(tenantId, System.currentTimeMillis() + 1, batchSize)).isEqualTo(1); } + @Test + public void syncCreateThenAsyncUpdateConvergesToUpdatedStatus() throws Exception { + UUID id = UUID.randomUUID(); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + // Production create path: the QUEUED row is persisted synchronously (persist-before-send). + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); + + Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(stored).isNotNull(); + assertThat(stored.getStatus()).isEqualTo(RpcStatus.QUEUED); + assertThat(stored.getResponse()).isNull(); + + // The async status update matches the existing row, so the future resolves true. + assertThat(rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.DELIVERED, JacksonUtil.toJsonNode("{\"ok\":true}"))) + .get(5, TimeUnit.SECONDS)).isTrue(); + + Rpc afterUpdate = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(afterUpdate.getStatus()).isEqualTo(RpcStatus.DELIVERED); + assertThat(afterUpdate.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); + } + + @Test + public void updateBatchAppliesInOrderAndAlignsResults() { + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + UUID idA = UUID.randomUUID(); + UUID idB = UUID.randomUUID(); // never created -> its update must report no match + + // Row A exists (persisted synchronously at create time); B was never created. + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(idA, deviceId, RpcStatus.QUEUED, null)); + + // Drive the persist logic directly with a single, deterministically-coalesced update batch. + // This is exactly what "coalescing" means: one update() call carrying several writes for the + // same partition in submission order. No queue timing involved. + // index 0: update A (SUCCESSFUL, {ok:true}) -> UPDATE hits the existing row -> true + // index 1: update B (SUCCESSFUL, {x:1}) -> UPDATE for a missing row -> false + // index 2: update A (EXPIRED, null response) -> UPDATE hits A, keeps response -> true + List batch = List.of( + new RpcEntity(rpc(idA, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))), + new RpcEntity(rpc(idB, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"x\":1}"))), + new RpcEntity(rpc(idA, deviceId, RpcStatus.EXPIRED, null))); + + List persisted = rpcUpdateRepository.update(batch); + + // Booleans are aligned positionally to submission order. + assertThat(persisted).containsExactly(true, false, true); + + // Updates apply in order, so A ends EXPIRED, and the null-response update kept the stored response. + Rpc storedA = rpcDao.findById(TenantId.SYS_TENANT_ID, idA); + assertThat(storedA).isNotNull(); + assertThat(storedA.getStatus()).isEqualTo(RpcStatus.EXPIRED); + assertThat(storedA.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); + // ...and the update for a never-created row neither persisted nor resurrected it. + assertThat(rpcDao.findById(TenantId.SYS_TENANT_ID, idB)).isNull(); + } + + @Test + public void saveAsyncRequeueUpdatesExistingRowToQueued() throws Exception { + UUID id = UUID.randomUUID(); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + // Initial create. + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); + + // Delivery timeout with closeTransportSessionOnRpcDeliveryTimeout=true re-queues the RPC: the + // device actor persists status=QUEUED again as a status update so init() can re-pick it up. + // The update must land on the existing row, never be dropped. + rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + + Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(stored).isNotNull(); + assertThat(stored.getStatus()).isEqualTo(RpcStatus.QUEUED); + } + + @Test + public void saveAsyncNullResponseUpdateKeepsStoredResponse() throws Exception { + UUID id = UUID.randomUUID(); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); + + // A successful response is stored. + rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))) + .get(5, TimeUnit.SECONDS); + + // A later status update carries no response - it must NOT clobber the stored one. + rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.EXPIRED, null)).get(5, TimeUnit.SECONDS); + + Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(stored.getStatus()).isEqualTo(RpcStatus.EXPIRED); + assertThat(stored.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); + } + + @Test + public void saveAsyncUpdateForDeletedRpcDoesNotResurrect() throws Exception { + UUID id = UUID.randomUUID(); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); + + // RPC is removed (TTL cleanup / manual delete) while a response is still in flight. + rpcDao.removeById(TenantId.SYS_TENANT_ID, id); + assertThat(rpcDao.findById(TenantId.SYS_TENANT_ID, id)).isNull(); + + // A late status update must not re-create the deleted row, and the future must resolve false + // (no row matched) so the service layer can skip the rule-engine notification. + assertThat(rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))) + .get(5, TimeUnit.SECONDS)).isFalse(); + + assertThat(rpcDao.findById(TenantId.SYS_TENANT_ID, id)).isNull(); + } + + private Rpc rpc(UUID id, DeviceId deviceId, RpcStatus status, JsonNode response) { + Rpc rpc = new Rpc(new RpcId(id)); + rpc.setCreatedTime(System.currentTimeMillis()); + rpc.setTenantId(TenantId.SYS_TENANT_ID); + rpc.setDeviceId(deviceId); + rpc.setExpirationTime(System.currentTimeMillis() + 60_000); + rpc.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}")); + rpc.setStatus(status); + rpc.setResponse(response); + return rpc; + } + }