From 0f41970cc76c012fd63184321abdccffcdca9b6c Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 25 Jun 2026 16:51:06 +0300 Subject: [PATCH 01/12] feat(rpc): batched async upsert persistence in RPC DAO --- .../src/main/resources/thingsboard.yml | 6 ++ .../server/dao/rpc/RpcService.java | 2 + .../server/dao/rpc/BaseRpcService.java | 6 ++ .../thingsboard/server/dao/rpc/RpcDao.java | 3 + .../server/dao/sql/rpc/JpaRpcDao.java | 64 +++++++++++++++++- .../dao/sql/rpc/RpcInsertRepository.java | 66 +++++++++++++++++++ .../server/dao/sql/rpc/JpaRpcDaoTest.java | 36 ++++++++++ 7 files changed, 180 insertions(+), 3 deletions(-) create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index f678076c57..3209c54746 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -424,6 +424,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 inserts/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}" # batch thread count has to be a prime number like 3 or 5 to gain perfect hash distribution + callback_threads: "${SQL_RPC_CALLBACK_THREADS:3}" # striped threads for post-persist rule-engine notifications; keep equal to batch_threads for aligned ordering 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/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..deb3bf8cf6 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 saveAsync(Rpc rpc); + void deleteRpc(TenantId tenantId, RpcId id); void deleteAllRpcByTenantId(TenantId tenantId); 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..4162422e69 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 saveAsync(Rpc rpc) { + log.trace("Executing saveAsync, [{}]", rpc); + return rpcDao.saveAsync(rpc.getTenantId(), 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..a823b05507 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 saveAsync(TenantId tenantId, 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/rpc/JpaRpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java index a929b6d1e4..e46dc2a704 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.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +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,76 @@ 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 public class JpaRpcDao extends JpaAbstractDao implements RpcDao, TenantEntityDao { - private final RpcRepository rpcRepository; + @Autowired + private RpcRepository rpcRepository; + @Autowired + private RpcInsertRepository rpcInsertRepository; + @Autowired + private ScheduledLogExecutorComponent logExecutor; + @Autowired + private 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(false) + .build(); + Function hashcodeFunction = entity -> entity.getUuid().hashCode(); + queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory); + queue.init(logExecutor, entities -> rpcInsertRepository.saveOrUpdate(entities), + Comparator.comparing(RpcEntity::getUuid)); + } + + @PreDestroy + private void destroy() { + if (queue != null) { + queue.destroy(); + } + } + + @Override + public ListenableFuture saveAsync(TenantId tenantId, 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/RpcInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java new file mode 100644 index 0000000000..5073702ef4 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java @@ -0,0 +1,66 @@ +/** + * 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.List; + +@Repository +public class RpcInsertRepository extends AbstractInsertRepository { + + private static final String INSERT_OR_UPDATE = + "INSERT INTO rpc (id, created_time, tenant_id, device_id, expiration_time, request, response, additional_info, status) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, response = EXCLUDED.response;"; + + public void saveOrUpdate(List entities) { + transactionTemplate.execute(status -> { + jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + RpcEntity rpc = entities.get(i); + ps.setObject(1, rpc.getUuid()); + ps.setLong(2, rpc.getCreatedTime()); + ps.setObject(3, rpc.getTenantId()); + ps.setObject(4, rpc.getDeviceId()); + ps.setLong(5, rpc.getExpirationTime()); + ps.setString(6, toJsonStr(rpc.getRequest())); + ps.setString(7, toJsonStr(rpc.getResponse())); + ps.setString(8, toJsonStr(rpc.getAdditionalInfo())); + ps.setString(9, rpc.getStatus().name()); + } + + @Override + public int getBatchSize() { + return entities.size(); + } + }); + return null; + }); + } + + 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..8b95aa7c3b 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 @@ -19,12 +19,14 @@ 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 java.util.UUID; +import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; @@ -57,4 +59,38 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { assertThat(rpcDao.deleteOutdatedRpcByTenantIdBatch(tenantId, System.currentTimeMillis() + 1, batchSize)).isEqualTo(1); } + @Test + public void saveAsyncInsertThenUpsert() throws Exception { + UUID id = UUID.randomUUID(); + Rpc rpc = new Rpc(new RpcId(id)); + rpc.setCreatedTime(System.currentTimeMillis()); + rpc.setTenantId(TenantId.SYS_TENANT_ID); + rpc.setDeviceId(new DeviceId(UUID.randomUUID())); + rpc.setExpirationTime(System.currentTimeMillis() + 60_000); + rpc.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}")); + rpc.setStatus(RpcStatus.QUEUED); + + rpcDao.saveAsync(rpc.getTenantId(), rpc).get(5, TimeUnit.SECONDS); + + Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(stored).isNotNull(); + assertThat(stored.getStatus()).isEqualTo(RpcStatus.QUEUED); + assertThat(stored.getResponse()).isNull(); + + Rpc update = new Rpc(new RpcId(id)); + update.setCreatedTime(rpc.getCreatedTime()); + update.setTenantId(TenantId.SYS_TENANT_ID); + update.setDeviceId(rpc.getDeviceId()); + update.setExpirationTime(rpc.getExpirationTime()); + update.setRequest(rpc.getRequest()); + update.setStatus(RpcStatus.DELIVERED); + update.setResponse(JacksonUtil.toJsonNode("{\"ok\":true}")); + + rpcDao.saveAsync(update.getTenantId(), update).get(5, TimeUnit.SECONDS); + + Rpc afterUpdate = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(afterUpdate.getStatus()).isEqualTo(RpcStatus.DELIVERED); + assertThat(afterUpdate.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); + } + } From 58d5a44c1dea06e51bb4ed02ef5151fc8c0d6ede Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 25 Jun 2026 17:08:58 +0300 Subject: [PATCH 02/12] feat(rpc): async RPC persistence with striped post-flush rule-engine notification --- .../server/service/rpc/TbRpcService.java | 49 +++++++++++-- .../server/service/rpc/TbRpcServiceTest.java | 70 +++++++++++++++++++ 2 files changed, 115 insertions(+), 4 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java 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..60ace279ba 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 @@ -16,10 +16,16 @@ package org.thingsboard.server.service.rpc; import com.fasterxml.jackson.databind.JsonNode; +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.stereotype.Service; +import org.thingsboard.common.util.DonAsynchron; 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; @@ -34,6 +40,11 @@ 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 @@ -42,10 +53,40 @@ 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; + @Value("${sql.rpc.callback_threads:3}") + private int callbackThreads; + + private ExecutorService[] callbackExecutors; + + @PostConstruct + private void init() { + callbackExecutors = new ExecutorService[callbackThreads]; + for (int i = 0; i < callbackThreads; i++) { + callbackExecutors[i] = Executors.newSingleThreadExecutor( + ThingsBoardThreadFactory.forName("rpc-persist-callback-" + i)); + } + } + + @PreDestroy + private void destroy() { + if (callbackExecutors != null) { + for (ExecutorService executor : callbackExecutors) { + executor.shutdownNow(); + } + } + } + + public void save(TenantId tenantId, Rpc rpc) { + ListenableFuture future = rpcService.saveAsync(rpc); + DonAsynchron.withCallback(future, + v -> pushRpcMsgToRuleEngine(tenantId, rpc), + 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[(rpcId.hashCode() & 0x7FFFFFFF) % callbackThreads]; } public void save(TenantId tenantId, RpcId rpcId, RpcStatus newStatus, JsonNode response) { 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..d0736ada26 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -0,0 +1,70 @@ +/** + * 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.springframework.test.util.ReflectionTestUtils; +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.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.UUID; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +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); + ReflectionTestUtils.setField(tbRpcService, "callbackThreads", 1); + ReflectionTestUtils.invokeMethod(tbRpcService, "init"); + } + + @Test + public void savePushesToRuleEngineAfterFlush() { + Rpc rpc = new Rpc(new RpcId(UUID.randomUUID())); + rpc.setTenantId(TenantId.SYS_TENANT_ID); + rpc.setDeviceId(new DeviceId(UUID.randomUUID())); + rpc.setStatus(RpcStatus.QUEUED); + rpc.setRequest(JacksonUtil.toJsonNode("{}")); + + when(rpcService.saveAsync(rpc)).thenReturn(Futures.immediateFuture(null)); + + tbRpcService.save(rpc.getTenantId(), rpc); + + verify(clusterService, timeout(5000)) + .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); + } +} From 134b2012bd63e157cc5c4643fe6d1c5f17e0551c Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 09:57:52 +0300 Subject: [PATCH 03/12] refactor(rpc): device actor builds full Rpc for async upsert at all save sites --- .../device/DeviceActorMessageProcessor.java | 15 ++++++++++----- .../server/service/rpc/TbRpcService.java | 15 --------------- 2 files changed, 10 insertions(+), 20 deletions(-) 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..ef48e6791b 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 @@ -258,15 +258,20 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } private void createRpc(ToDeviceRpcRequest request, RpcStatus status) { + systemContext.getTbRpcService().save(tenantId, buildRpc(request, status, null)); + } + + private Rpc buildRpc(ToDeviceRpcRequest request, RpcStatus status, JsonNode response) { Rpc rpc = new Rpc(new RpcId(request.getId())); rpc.setCreatedTime(System.currentTimeMillis()); 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) { @@ -351,7 +356,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().save(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.EXPIRED, null)); } systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION)); @@ -662,7 +667,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().save(tenantId, buildRpc(toDeviceRequestMsg, status, response)); } } finally { if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE)) { @@ -726,7 +731,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } if (persisted) { - systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, response); + systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRpcRequest, status, response)); } if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE) && status.equals(RpcStatus.DELIVERED) && !oneWayRpc) { @@ -819,7 +824,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().save(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout)); } }, systemContext.getRpcResponseTimeout(), TimeUnit.MILLISECONDS); } 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 60ace279ba..2f5c5ea6a8 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,7 +15,6 @@ */ package org.thingsboard.server.service.rpc; -import com.fasterxml.jackson.databind.JsonNode; import com.google.common.util.concurrent.ListenableFuture; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; @@ -89,20 +88,6 @@ public class TbRpcService { return callbackExecutors[(rpcId.hashCode() & 0x7FFFFFFF) % callbackThreads]; } - 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); - } - } - private void pushRpcMsgToRuleEngine(TenantId tenantId, Rpc rpc) { TbMsg msg = TbMsg.newMsg() .type(TbMsgType.valueOf("RPC_" + rpc.getStatus().name())) From e0ebd349c53513410087ae3a19c594e590c95f17 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 10:15:13 +0300 Subject: [PATCH 04/12] test(rpc): cover same-rpcId coalesced-batch ordering; clarify callback_threads comment --- .../src/main/resources/thingsboard.yml | 2 +- .../server/dao/sql/rpc/JpaRpcDaoTest.java | 38 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 3209c54746..f498d222ac 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -429,7 +429,7 @@ sql: 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}" # batch thread count has to be a prime number like 3 or 5 to gain perfect hash distribution - callback_threads: "${SQL_RPC_CALLBACK_THREADS:3}" # striped threads for post-persist rule-engine notifications; keep equal to batch_threads for aligned ordering + 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 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/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 8b95aa7c3b..f631ff2daf 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 @@ -93,4 +93,42 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { assertThat(afterUpdate.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); } + @Test + public void saveAsyncSameRpcIdCoalescedBatchKeepsLatestStatus() throws Exception { + UUID id = UUID.randomUUID(); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + long createdTime = System.currentTimeMillis(); + long expirationTime = createdTime + 60_000; + + Rpc queued = new Rpc(new RpcId(id)); + queued.setCreatedTime(createdTime); + queued.setTenantId(TenantId.SYS_TENANT_ID); + queued.setDeviceId(deviceId); + queued.setExpirationTime(expirationTime); + queued.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}")); + queued.setStatus(RpcStatus.QUEUED); + + Rpc delivered = new Rpc(new RpcId(id)); + delivered.setCreatedTime(createdTime); + delivered.setTenantId(TenantId.SYS_TENANT_ID); + delivered.setDeviceId(deviceId); + delivered.setExpirationTime(expirationTime); + delivered.setRequest(queued.getRequest()); + delivered.setStatus(RpcStatus.DELIVERED); + delivered.setResponse(JacksonUtil.toJsonNode("{\"ok\":true}")); + + // Enqueue both writes for the same rpcId back-to-back so they coalesce into one flush batch. + // Same rpcId -> same partition; the queue's stable sort must keep submission order + // (QUEUED before DELIVERED), so the final persisted row must be DELIVERED, never QUEUED. + var queuedFuture = rpcDao.saveAsync(TenantId.SYS_TENANT_ID, queued); + var deliveredFuture = rpcDao.saveAsync(TenantId.SYS_TENANT_ID, delivered); + queuedFuture.get(5, TimeUnit.SECONDS); + deliveredFuture.get(5, TimeUnit.SECONDS); + + Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); + assertThat(stored).isNotNull(); + assertThat(stored.getStatus()).isEqualTo(RpcStatus.DELIVERED); + assertThat(stored.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); + } + } From f8d59a25e412312986e2e42d59a9857369c3933c Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 15:59:51 +0300 Subject: [PATCH 05/12] fix(rpc): preserve response and prevent row resurrection in batched upsert Split the batched RPC persistence into a create path (INSERT ... ON CONFLICT) and an update path (UPDATE ... WHERE id = ?). A status update for a row deleted in the meantime (TTL cleanup / manual delete) is no longer resurrected, restoring the old findById-null skip. Both paths COALESCE the response so a status update that carries none no longer clobbers a previously stored one. The create-vs-update intent is carried by a dedicated RpcQueueEntry record (keeping RpcEntity pure data) and exposed via explicit createAsync/updateAsync DAO methods instead of a boolean flag; insert/update batch scaffolding is collapsed into a shared helper. Adds DAO regression tests (re-queue keeps existing row, null-response preservation, no resurrection of a deleted row) and TbRpcService create/update wiring tests. --- .../device/DeviceActorMessageProcessor.java | 2 +- .../server/service/rpc/TbRpcService.java | 9 ++- .../server/service/rpc/TbRpcServiceTest.java | 33 ++++++-- .../server/dao/rpc/RpcService.java | 4 +- .../server/dao/rpc/BaseRpcService.java | 12 ++- .../thingsboard/server/dao/rpc/RpcDao.java | 4 +- .../server/dao/sql/rpc/JpaRpcDao.java | 17 +++-- .../dao/sql/rpc/RpcInsertRepository.java | 60 +++++++++++---- .../server/dao/sql/rpc/RpcQueueEntry.java | 34 +++++++++ .../server/dao/sql/rpc/JpaRpcDaoTest.java | 76 ++++++++++++++++++- 10 files changed, 212 insertions(+), 39 deletions(-) create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java 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 ef48e6791b..6f257405af 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 @@ -258,7 +258,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } private void createRpc(ToDeviceRpcRequest request, RpcStatus status) { - systemContext.getTbRpcService().save(tenantId, buildRpc(request, status, null)); + systemContext.getTbRpcService().create(tenantId, buildRpc(request, status, null)); } private Rpc buildRpc(ToDeviceRpcRequest request, RpcStatus status, JsonNode response) { 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 2f5c5ea6a8..486feab2e2 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 @@ -75,8 +75,15 @@ public class TbRpcService { } } + public void create(TenantId tenantId, Rpc rpc) { + persist(tenantId, rpc, rpcService.createAsync(rpc)); + } + public void save(TenantId tenantId, Rpc rpc) { - ListenableFuture future = rpcService.saveAsync(rpc); + persist(tenantId, rpc, rpcService.updateAsync(rpc)); + } + + private void persist(TenantId tenantId, Rpc rpc, ListenableFuture future) { DonAsynchron.withCallback(future, v -> pushRpcMsgToRuleEngine(tenantId, rpc), t -> log.error("[{}][{}][{}] Failed to persist RPC with status [{}]", 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 index d0736ada26..014ab256af 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -53,18 +53,35 @@ public class TbRpcServiceTest { } @Test - public void savePushesToRuleEngineAfterFlush() { - Rpc rpc = new Rpc(new RpcId(UUID.randomUUID())); - rpc.setTenantId(TenantId.SYS_TENANT_ID); - rpc.setDeviceId(new DeviceId(UUID.randomUUID())); - rpc.setStatus(RpcStatus.QUEUED); - rpc.setRequest(JacksonUtil.toJsonNode("{}")); - - when(rpcService.saveAsync(rpc)).thenReturn(Futures.immediateFuture(null)); + public void savePersistsViaUpdateAsyncThenPushesToRuleEngine() { + Rpc rpc = newRpc(); + when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(null)); tbRpcService.save(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 createPersistsViaCreateAsyncThenPushesToRuleEngine() { + Rpc rpc = newRpc(); + when(rpcService.createAsync(rpc)).thenReturn(Futures.immediateFuture(null)); + + tbRpcService.create(rpc.getTenantId(), rpc); + + verify(rpcService).createAsync(rpc); verify(clusterService, timeout(5000)) .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); } + + private Rpc newRpc() { + Rpc rpc = new Rpc(new RpcId(UUID.randomUUID())); + rpc.setTenantId(TenantId.SYS_TENANT_ID); + rpc.setDeviceId(new DeviceId(UUID.randomUUID())); + rpc.setStatus(RpcStatus.QUEUED); + 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 deb3bf8cf6..9dc821fcc7 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,7 +29,9 @@ public interface RpcService extends EntityDaoService { Rpc save(Rpc rpc); - ListenableFuture saveAsync(Rpc rpc); + ListenableFuture createAsync(Rpc rpc); + + ListenableFuture updateAsync(Rpc rpc); void deleteRpc(TenantId tenantId, RpcId id); 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 4162422e69..e374aaf9ce 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 @@ -55,9 +55,15 @@ public class BaseRpcService implements RpcService { } @Override - public ListenableFuture saveAsync(Rpc rpc) { - log.trace("Executing saveAsync, [{}]", rpc); - return rpcDao.saveAsync(rpc.getTenantId(), rpc); + public ListenableFuture createAsync(Rpc rpc) { + log.trace("Executing createAsync, [{}]", rpc); + return rpcDao.createAsync(rpc.getTenantId(), rpc); + } + + @Override + public ListenableFuture updateAsync(Rpc rpc) { + log.trace("Executing updateAsync, [{}]", rpc); + return rpcDao.updateAsync(rpc.getTenantId(), rpc); } @Override 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 a823b05507..7bf4f5f8ac 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 @@ -26,7 +26,9 @@ import org.thingsboard.server.dao.Dao; public interface RpcDao extends Dao { - ListenableFuture saveAsync(TenantId tenantId, Rpc rpc); + ListenableFuture createAsync(TenantId tenantId, Rpc rpc); + + ListenableFuture updateAsync(TenantId tenantId, Rpc rpc); PageData findAllByDeviceId(TenantId tenantId, DeviceId deviceId, PageLink pageLink); 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 e46dc2a704..45c765e42f 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 @@ -71,7 +71,7 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; - private TbSqlBlockingQueueWrapper queue; + private TbSqlBlockingQueueWrapper queue; @PostConstruct private void init() { @@ -84,10 +84,10 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, .batchSortEnabled(batchSortEnabled) .withResponse(false) .build(); - Function hashcodeFunction = entity -> entity.getUuid().hashCode(); + Function hashcodeFunction = entry -> entry.entity().getUuid().hashCode(); queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory); - queue.init(logExecutor, entities -> rpcInsertRepository.saveOrUpdate(entities), - Comparator.comparing(RpcEntity::getUuid)); + queue.init(logExecutor, entries -> rpcInsertRepository.saveOrUpdate(entries), + Comparator.comparing((RpcQueueEntry entry) -> entry.entity().getUuid())); } @PreDestroy @@ -98,8 +98,13 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, } @Override - public ListenableFuture saveAsync(TenantId tenantId, Rpc rpc) { - return queue.add(new RpcEntity(rpc)); + public ListenableFuture createAsync(TenantId tenantId, Rpc rpc) { + return queue.add(RpcQueueEntry.forInsert(new RpcEntity(rpc))); + } + + @Override + public ListenableFuture updateAsync(TenantId tenantId, Rpc rpc) { + return queue.add(RpcQueueEntry.forUpdate(new RpcEntity(rpc))); } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java index 5073702ef4..f3023b0c7c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java @@ -29,17 +29,33 @@ import java.util.List; @Repository public class RpcInsertRepository extends AbstractInsertRepository { - private static final String INSERT_OR_UPDATE = + // Used only for the initial persistence of a new RPC (RpcStatus.QUEUED / already-EXPIRED on arrival). + // ON CONFLICT keeps the create idempotent (e.g. on actor re-processing); COALESCE never clobbers an + // existing response with NULL. + private static final String INSERT = "INSERT INTO rpc (id, created_time, tenant_id, device_id, expiration_time, request, response, additional_info, status) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " + - "ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, response = EXCLUDED.response;"; + "ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, response = COALESCE(EXCLUDED.response, rpc.response);"; - public void saveOrUpdate(List entities) { + // Used for every subsequent status change. WHERE id = ? means a row deleted in the meantime + // (TTL cleanup / manual delete) is NOT resurrected - it matches the old findById-null skip. + // COALESCE preserves a previously stored response when a status update carries none. + private static final String UPDATE = + "UPDATE rpc SET status = ?, response = COALESCE(?, response) WHERE id = ?;"; + + @FunctionalInterface + private interface ColumnBinder { + void bind(PreparedStatement ps, RpcEntity rpc) throws SQLException; + } + + public void saveOrUpdate(List entries) { + List inserts = entries.stream().filter(RpcQueueEntry::insert).map(RpcQueueEntry::entity).toList(); + List updates = entries.stream().filter(entry -> !entry.insert()).map(RpcQueueEntry::entity).toList(); transactionTemplate.execute(status -> { - jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() { - @Override - public void setValues(PreparedStatement ps, int i) throws SQLException { - RpcEntity rpc = entities.get(i); + // Inserts run first so a create and a status update for the same rpcId coalesced into one + // batch still apply in create -> update order. + if (!inserts.isEmpty()) { + batch(INSERT, inserts, (ps, rpc) -> { ps.setObject(1, rpc.getUuid()); ps.setLong(2, rpc.getCreatedTime()); ps.setObject(3, rpc.getTenantId()); @@ -49,17 +65,33 @@ public class RpcInsertRepository extends AbstractInsertRepository { ps.setString(7, toJsonStr(rpc.getResponse())); ps.setString(8, toJsonStr(rpc.getAdditionalInfo())); ps.setString(9, rpc.getStatus().name()); - } - - @Override - public int getBatchSize() { - return entities.size(); - } - }); + }); + } + if (!updates.isEmpty()) { + batch(UPDATE, updates, (ps, rpc) -> { + ps.setString(1, rpc.getStatus().name()); + ps.setString(2, toJsonStr(rpc.getResponse())); + ps.setObject(3, rpc.getUuid()); + }); + } return null; }); } + private void batch(String sql, List entities, ColumnBinder binder) { + jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + binder.bind(ps, entities.get(i)); + } + + @Override + public int getBatchSize() { + return entities.size(); + } + }); + } + private String toJsonStr(JsonNode node) { return node == null ? null : replaceNullChars(JacksonUtil.toString(node)); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java new file mode 100644 index 0000000000..8b5a5f96cf --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java @@ -0,0 +1,34 @@ +/** + * 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 org.thingsboard.server.dao.model.sql.RpcEntity; + +/** + * Pairs an {@link RpcEntity} with the write intent for the batched persistence queue, so the entity + * itself stays pure data. {@code insert} selects INSERT-on-conflict (initial create) vs UPDATE-by-id + * (status change that must never resurrect a deleted row). + */ +record RpcQueueEntry(RpcEntity entity, boolean insert) { + + static RpcQueueEntry forInsert(RpcEntity entity) { + return new RpcQueueEntry(entity, true); + } + + static RpcQueueEntry forUpdate(RpcEntity entity) { + return new RpcQueueEntry(entity, false); + } +} 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 f631ff2daf..d49e0c2355 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,6 +15,7 @@ */ 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; @@ -70,7 +71,7 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { rpc.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}")); rpc.setStatus(RpcStatus.QUEUED); - rpcDao.saveAsync(rpc.getTenantId(), rpc).get(5, TimeUnit.SECONDS); + rpcDao.createAsync(rpc.getTenantId(), rpc).get(5, TimeUnit.SECONDS); Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); assertThat(stored).isNotNull(); @@ -86,7 +87,7 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { update.setStatus(RpcStatus.DELIVERED); update.setResponse(JacksonUtil.toJsonNode("{\"ok\":true}")); - rpcDao.saveAsync(update.getTenantId(), update).get(5, TimeUnit.SECONDS); + rpcDao.updateAsync(update.getTenantId(), update).get(5, TimeUnit.SECONDS); Rpc afterUpdate = rpcDao.findById(TenantId.SYS_TENANT_ID, id); assertThat(afterUpdate.getStatus()).isEqualTo(RpcStatus.DELIVERED); @@ -120,8 +121,8 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { // Enqueue both writes for the same rpcId back-to-back so they coalesce into one flush batch. // Same rpcId -> same partition; the queue's stable sort must keep submission order // (QUEUED before DELIVERED), so the final persisted row must be DELIVERED, never QUEUED. - var queuedFuture = rpcDao.saveAsync(TenantId.SYS_TENANT_ID, queued); - var deliveredFuture = rpcDao.saveAsync(TenantId.SYS_TENANT_ID, delivered); + var queuedFuture = rpcDao.createAsync(TenantId.SYS_TENANT_ID, queued); + var deliveredFuture = rpcDao.updateAsync(TenantId.SYS_TENANT_ID, delivered); queuedFuture.get(5, TimeUnit.SECONDS); deliveredFuture.get(5, TimeUnit.SECONDS); @@ -131,4 +132,71 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { assertThat(stored.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); } + @Test + public void saveAsyncRequeueUpdatesExistingRowToQueued() throws Exception { + UUID id = UUID.randomUUID(); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + // Initial create. + rpcDao.createAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + + // 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(TenantId.SYS_TENANT_ID, 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.createAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + + // A successful response is stored. + rpcDao.updateAsync(TenantId.SYS_TENANT_ID, 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(TenantId.SYS_TENANT_ID, 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.createAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + + // 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. + rpcDao.updateAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))) + .get(5, TimeUnit.SECONDS); + + 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; + } + } From ef571175f742ecad9ad6fc491891c9006e6ce504 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 17:04:24 +0300 Subject: [PATCH 06/12] fix(rpc): skip rule-engine notify when async update matches no row Async RPC persistence now returns a per-write Boolean: an INSERT-on-conflict always persists (true), while an UPDATE-by-id reports false when its WHERE id = ? matched no row (RPC deleted via TTL/manual delete). TbRpcService notifies the rule engine only when the write actually persisted, restoring the findById-null skip the async refactor had dropped. Also: drop the unused tenantId param from createAsync/updateAsync; make TbRpcService's callback-thread count constructor-injectable (removes test reflection) with a comment on why callback striping exists alongside the queue partitioning; correct the RPC batch_threads yml comment; assert the Boolean contract in JpaRpcDaoTest and add a notification-suppression unit test. --- .../server/service/rpc/TbRpcService.java | 41 +++++++++------- .../src/main/resources/thingsboard.yml | 2 +- .../server/service/rpc/TbRpcServiceTest.java | 24 +++++++--- .../server/dao/rpc/RpcService.java | 4 +- .../server/dao/rpc/BaseRpcService.java | 8 ++-- .../thingsboard/server/dao/rpc/RpcDao.java | 4 +- .../server/dao/sql/rpc/JpaRpcDao.java | 14 ++++-- .../dao/sql/rpc/RpcInsertRepository.java | 24 +++++++--- .../server/dao/sql/rpc/JpaRpcDaoTest.java | 47 +++++++------------ 9 files changed, 96 insertions(+), 72 deletions(-) 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 486feab2e2..da90db7846 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 @@ -16,9 +16,7 @@ package org.thingsboard.server.service.rpc; 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.stereotype.Service; @@ -46,20 +44,24 @@ import java.util.concurrent.Executors; @TbCoreComponent @Service -@RequiredArgsConstructor @Slf4j public class TbRpcService { private final RpcService rpcService; private final TbClusterService tbClusterService; - @Value("${sql.rpc.callback_threads:3}") - private int callbackThreads; + // Post-persist rule-engine notifications run on these striped single-thread executors, keyed by + // rpcId, instead of inline on the SQL persist threads. A flush batch can carry up to batch_size + // entries, and each notification's pushMsgToRuleEngine may do a (potentially blocking) device-profile + // cache lookup; running them inline would serialize that work on the persist thread and stall the + // next DB flush. Striping by rpcId (rather than a shared multi-threaded pool) also preserves + // per-command notification order, e.g. RPC_QUEUED before RPC_DELIVERED. + private final ExecutorService[] callbackExecutors; - private ExecutorService[] callbackExecutors; - - @PostConstruct - private void init() { - callbackExecutors = new ExecutorService[callbackThreads]; + public TbRpcService(RpcService rpcService, TbClusterService tbClusterService, + @Value("${sql.rpc.callback_threads:3}") int 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)); @@ -68,10 +70,8 @@ public class TbRpcService { @PreDestroy private void destroy() { - if (callbackExecutors != null) { - for (ExecutorService executor : callbackExecutors) { - executor.shutdownNow(); - } + for (ExecutorService executor : callbackExecutors) { + executor.shutdownNow(); } } @@ -83,16 +83,23 @@ public class TbRpcService { persist(tenantId, rpc, rpcService.updateAsync(rpc)); } - private void persist(TenantId tenantId, Rpc rpc, ListenableFuture future) { + private void persist(TenantId tenantId, Rpc rpc, ListenableFuture future) { DonAsynchron.withCallback(future, - v -> pushRpcMsgToRuleEngine(tenantId, rpc), + persisted -> { + if (Boolean.TRUE.equals(persisted)) { + pushRpcMsgToRuleEngine(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[(rpcId.hashCode() & 0x7FFFFFFF) % callbackThreads]; + return callbackExecutors[(rpcId.hashCode() & 0x7FFFFFFF) % callbackExecutors.length]; } private void pushRpcMsgToRuleEngine(TenantId tenantId, Rpc rpc) { diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index f498d222ac..19075e7305 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -428,7 +428,7 @@ sql: batch_size: "${SQL_RPC_BATCH_SIZE:1000}" # Batch size for persisting RPC inserts/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}" # batch thread count has to be a prime number like 3 or 5 to gain perfect hash distribution + 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 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 events: batch_size: "${SQL_EVENTS_BATCH_SIZE:10000}" # Batch size for persisting latest telemetry updates 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 index 014ab256af..87162c1123 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -18,7 +18,6 @@ package org.thingsboard.server.service.rpc; import com.google.common.util.concurrent.Futures; import org.junit.Before; import org.junit.Test; -import org.springframework.test.util.ReflectionTestUtils; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.id.DeviceId; @@ -34,6 +33,7 @@ import java.util.UUID; 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.mock; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; @@ -47,15 +47,13 @@ public class TbRpcServiceTest { @Before public void setUp() { - tbRpcService = new TbRpcService(rpcService, clusterService); - ReflectionTestUtils.setField(tbRpcService, "callbackThreads", 1); - ReflectionTestUtils.invokeMethod(tbRpcService, "init"); + tbRpcService = new TbRpcService(rpcService, clusterService, 1); } @Test public void savePersistsViaUpdateAsyncThenPushesToRuleEngine() { Rpc rpc = newRpc(); - when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(null)); + when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(true)); tbRpcService.save(rpc.getTenantId(), rpc); @@ -67,7 +65,7 @@ public class TbRpcServiceTest { @Test public void createPersistsViaCreateAsyncThenPushesToRuleEngine() { Rpc rpc = newRpc(); - when(rpcService.createAsync(rpc)).thenReturn(Futures.immediateFuture(null)); + when(rpcService.createAsync(rpc)).thenReturn(Futures.immediateFuture(true)); tbRpcService.create(rpc.getTenantId(), rpc); @@ -76,6 +74,20 @@ public class TbRpcServiceTest { .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); } + @Test + public void saveDoesNotNotifyRuleEngineWhenRowMissing() { + 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.save(rpc.getTenantId(), rpc); + + verify(rpcService).updateAsync(rpc); + verify(clusterService, after(500).never()) + .pushMsgToRuleEngine(any(TenantId.class), any(DeviceId.class), any(TbMsg.class), isNull()); + } + private Rpc newRpc() { Rpc rpc = new Rpc(new RpcId(UUID.randomUUID())); rpc.setTenantId(TenantId.SYS_TENANT_ID); 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 9dc821fcc7..aaac66c86b 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,9 +29,9 @@ public interface RpcService extends EntityDaoService { Rpc save(Rpc rpc); - ListenableFuture createAsync(Rpc rpc); + ListenableFuture createAsync(Rpc rpc); - ListenableFuture updateAsync(Rpc rpc); + ListenableFuture updateAsync(Rpc rpc); void deleteRpc(TenantId tenantId, RpcId id); 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 e374aaf9ce..759b90f3a2 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 @@ -55,15 +55,15 @@ public class BaseRpcService implements RpcService { } @Override - public ListenableFuture createAsync(Rpc rpc) { + public ListenableFuture createAsync(Rpc rpc) { log.trace("Executing createAsync, [{}]", rpc); - return rpcDao.createAsync(rpc.getTenantId(), rpc); + return rpcDao.createAsync(rpc); } @Override - public ListenableFuture updateAsync(Rpc rpc) { + public ListenableFuture updateAsync(Rpc rpc) { log.trace("Executing updateAsync, [{}]", rpc); - return rpcDao.updateAsync(rpc.getTenantId(), rpc); + return rpcDao.updateAsync(rpc); } @Override 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 7bf4f5f8ac..b0dde0d969 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 @@ -26,9 +26,9 @@ import org.thingsboard.server.dao.Dao; public interface RpcDao extends Dao { - ListenableFuture createAsync(TenantId tenantId, Rpc rpc); + ListenableFuture createAsync(Rpc rpc); - ListenableFuture updateAsync(TenantId tenantId, Rpc rpc); + ListenableFuture updateAsync(Rpc rpc); PageData findAllByDeviceId(TenantId tenantId, DeviceId deviceId, PageLink pageLink); 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 45c765e42f..d7fe95b3ca 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 @@ -71,7 +71,10 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; - private TbSqlBlockingQueueWrapper queue; + // Boolean response per queued write: an INSERT-on-conflict always persists (true), while an + // UPDATE-by-id reports false when it matched no row (the RPC was deleted in the meantime), so the + // service layer can skip the rule-engine notification for a row that no longer exists. + private TbSqlBlockingQueueWrapper queue; @PostConstruct private void init() { @@ -82,12 +85,13 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, .statsPrintIntervalMs(statsPrintIntervalMs) .statsNamePrefix("rpc") .batchSortEnabled(batchSortEnabled) - .withResponse(false) + .withResponse(true) .build(); Function hashcodeFunction = entry -> entry.entity().getUuid().hashCode(); queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory); queue.init(logExecutor, entries -> rpcInsertRepository.saveOrUpdate(entries), - Comparator.comparing((RpcQueueEntry entry) -> entry.entity().getUuid())); + Comparator.comparing((RpcQueueEntry entry) -> entry.entity().getUuid()), + Function.identity()); } @PreDestroy @@ -98,12 +102,12 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, } @Override - public ListenableFuture createAsync(TenantId tenantId, Rpc rpc) { + public ListenableFuture createAsync(Rpc rpc) { return queue.add(RpcQueueEntry.forInsert(new RpcEntity(rpc))); } @Override - public ListenableFuture updateAsync(TenantId tenantId, Rpc rpc) { + public ListenableFuture updateAsync(Rpc rpc) { return queue.add(RpcQueueEntry.forUpdate(new RpcEntity(rpc))); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java index f3023b0c7c..9d0eb45464 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java @@ -24,6 +24,7 @@ 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 @@ -48,10 +49,10 @@ public class RpcInsertRepository extends AbstractInsertRepository { void bind(PreparedStatement ps, RpcEntity rpc) throws SQLException; } - public void saveOrUpdate(List entries) { + public List saveOrUpdate(List entries) { List inserts = entries.stream().filter(RpcQueueEntry::insert).map(RpcQueueEntry::entity).toList(); List updates = entries.stream().filter(entry -> !entry.insert()).map(RpcQueueEntry::entity).toList(); - transactionTemplate.execute(status -> { + int[] updateCounts = transactionTemplate.execute(status -> { // Inserts run first so a create and a status update for the same rpcId coalesced into one // batch still apply in create -> update order. if (!inserts.isEmpty()) { @@ -68,18 +69,29 @@ public class RpcInsertRepository extends AbstractInsertRepository { }); } if (!updates.isEmpty()) { - batch(UPDATE, updates, (ps, rpc) -> { + return batch(UPDATE, updates, (ps, rpc) -> { ps.setString(1, rpc.getStatus().name()); ps.setString(2, toJsonStr(rpc.getResponse())); ps.setObject(3, rpc.getUuid()); }); } - return null; + return new int[0]; }); + + // Result is aligned to the submission order of entries: an insert always persists + // (INSERT ... ON CONFLICT), an update persists only if its WHERE id = ? matched a still-existing + // row. updateCounts keeps the same relative order as the filtered updates, so a single cursor + // walks it as we encounter update entries; insert entries short-circuit and don't advance it. + List persisted = new ArrayList<>(entries.size()); + int updateIdx = 0; + for (RpcQueueEntry entry : entries) { + persisted.add(entry.insert() || updateCounts[updateIdx++] > 0); + } + return persisted; } - private void batch(String sql, List entities, ColumnBinder binder) { - jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() { + private int[] batch(String sql, List entities, ColumnBinder binder) { + return jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() { @Override public void setValues(PreparedStatement ps, int i) throws SQLException { binder.bind(ps, entities.get(i)); 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 d49e0c2355..ad3f82c5bd 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 @@ -63,31 +63,19 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { @Test public void saveAsyncInsertThenUpsert() throws Exception { UUID id = UUID.randomUUID(); - Rpc rpc = new Rpc(new RpcId(id)); - rpc.setCreatedTime(System.currentTimeMillis()); - rpc.setTenantId(TenantId.SYS_TENANT_ID); - rpc.setDeviceId(new DeviceId(UUID.randomUUID())); - rpc.setExpirationTime(System.currentTimeMillis() + 60_000); - rpc.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}")); - rpc.setStatus(RpcStatus.QUEUED); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); - rpcDao.createAsync(rpc.getTenantId(), rpc).get(5, TimeUnit.SECONDS); + // A create always persists (INSERT ... ON CONFLICT), so the future resolves true. + assertThat(rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS)).isTrue(); Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); assertThat(stored).isNotNull(); assertThat(stored.getStatus()).isEqualTo(RpcStatus.QUEUED); assertThat(stored.getResponse()).isNull(); - Rpc update = new Rpc(new RpcId(id)); - update.setCreatedTime(rpc.getCreatedTime()); - update.setTenantId(TenantId.SYS_TENANT_ID); - update.setDeviceId(rpc.getDeviceId()); - update.setExpirationTime(rpc.getExpirationTime()); - update.setRequest(rpc.getRequest()); - update.setStatus(RpcStatus.DELIVERED); - update.setResponse(JacksonUtil.toJsonNode("{\"ok\":true}")); - - rpcDao.updateAsync(update.getTenantId(), update).get(5, TimeUnit.SECONDS); + // The 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); @@ -121,8 +109,8 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { // Enqueue both writes for the same rpcId back-to-back so they coalesce into one flush batch. // Same rpcId -> same partition; the queue's stable sort must keep submission order // (QUEUED before DELIVERED), so the final persisted row must be DELIVERED, never QUEUED. - var queuedFuture = rpcDao.createAsync(TenantId.SYS_TENANT_ID, queued); - var deliveredFuture = rpcDao.updateAsync(TenantId.SYS_TENANT_ID, delivered); + var queuedFuture = rpcDao.createAsync(queued); + var deliveredFuture = rpcDao.updateAsync(delivered); queuedFuture.get(5, TimeUnit.SECONDS); deliveredFuture.get(5, TimeUnit.SECONDS); @@ -138,12 +126,12 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { DeviceId deviceId = new DeviceId(UUID.randomUUID()); // Initial create. - rpcDao.createAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); // 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(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); assertThat(stored).isNotNull(); @@ -155,14 +143,14 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { UUID id = UUID.randomUUID(); DeviceId deviceId = new DeviceId(UUID.randomUUID()); - rpcDao.createAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); // A successful response is stored. - rpcDao.updateAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))) + 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(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.EXPIRED, null)).get(5, TimeUnit.SECONDS); + 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); @@ -174,15 +162,16 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { UUID id = UUID.randomUUID(); DeviceId deviceId = new DeviceId(UUID.randomUUID()); - rpcDao.createAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); // 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. - rpcDao.updateAsync(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))) - .get(5, TimeUnit.SECONDS); + // 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(); } From 879936b03f51bc6b32b33d77cb5f71a02ddc12a4 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 18:18:15 +0300 Subject: [PATCH 07/12] refactor(rpc): address review feedback on batched RPC persistence - Rename TbRpcService.save -> update so the create/update pairing is self-documenting (matches createAsync/updateAsync); update all actor call sites and unit tests. - Extract the (hash & 0x7FFFFFFF) % n striping into a shared HashPartitioner helper used by both TbSqlBlockingQueueWrapper and TbRpcService, so the end-to-end submission-order invariant has a single source of truth. - Collapse the insert/update split in RpcInsertRepository.saveOrUpdate into a single partitioningBy pass. - Remove the unused TbRpcService.findRpcById (and its now-orphaned import). - Restore the prime-thread-count guidance on the RPC batch_threads / callback_threads yml comments. --- .../device/DeviceActorMessageProcessor.java | 10 ++--- .../server/service/rpc/TbRpcService.java | 10 ++--- .../src/main/resources/thingsboard.yml | 4 +- .../server/service/rpc/TbRpcServiceTest.java | 8 ++-- .../common/util/HashPartitioner.java | 41 +++++++++++++++++++ .../dao/sql/TbSqlBlockingQueueWrapper.java | 3 +- .../dao/sql/rpc/RpcInsertRepository.java | 8 +++- 7 files changed, 63 insertions(+), 21 deletions(-) create mode 100644 common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java 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 6f257405af..6d0f07e507 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 @@ -356,7 +356,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, buildRpc(toDeviceRpcRequest, RpcStatus.EXPIRED, null)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.EXPIRED, null)); } systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION)); @@ -667,7 +667,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } catch (IllegalArgumentException e) { response = JacksonUtil.newObjectNode().put("error", payload); } - systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRequestMsg, status, response)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRequestMsg, status, response)); } } finally { if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE)) { @@ -731,7 +731,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } if (persisted) { - systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRpcRequest, status, response)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, status, response)); } if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE) && status.equals(RpcStatus.DELIVERED) && !oneWayRpc) { @@ -824,7 +824,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, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout)); } }, systemContext.getRpcResponseTimeout(), TimeUnit.MILLISECONDS); } @@ -1035,7 +1035,7 @@ 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); } 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 da90db7846..e2f1e00fb1 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 @@ -21,11 +21,11 @@ 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; @@ -79,7 +79,7 @@ public class TbRpcService { persist(tenantId, rpc, rpcService.createAsync(rpc)); } - public void save(TenantId tenantId, Rpc rpc) { + public void update(TenantId tenantId, Rpc rpc) { persist(tenantId, rpc, rpcService.updateAsync(rpc)); } @@ -99,7 +99,7 @@ public class TbRpcService { } private Executor executorFor(UUID rpcId) { - return callbackExecutors[(rpcId.hashCode() & 0x7FFFFFFF) % callbackExecutors.length]; + return callbackExecutors[HashPartitioner.resolvePartition(rpcId.hashCode(), callbackExecutors.length)]; } private void pushRpcMsgToRuleEngine(TenantId tenantId, Rpc rpc) { @@ -112,10 +112,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 19075e7305..af1b7d328d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -428,8 +428,8 @@ sql: batch_size: "${SQL_RPC_BATCH_SIZE:1000}" # Batch size for persisting RPC inserts/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 - 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 + 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. 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 index 87162c1123..cf11dfed48 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -51,11 +51,11 @@ public class TbRpcServiceTest { } @Test - public void savePersistsViaUpdateAsyncThenPushesToRuleEngine() { + public void updatePersistsViaUpdateAsyncThenPushesToRuleEngine() { Rpc rpc = newRpc(); when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(true)); - tbRpcService.save(rpc.getTenantId(), rpc); + tbRpcService.update(rpc.getTenantId(), rpc); verify(rpcService).updateAsync(rpc); verify(clusterService, timeout(5000)) @@ -75,13 +75,13 @@ public class TbRpcServiceTest { } @Test - public void saveDoesNotNotifyRuleEngineWhenRowMissing() { + 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.save(rpc.getTenantId(), rpc); + tbRpcService.update(rpc.getTenantId(), rpc); verify(rpcService).updateAsync(rpc); verify(clusterService, after(500).never()) 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..54555b2639 --- /dev/null +++ b/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java @@ -0,0 +1,41 @@ +/** + * 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)}. + *

+ * Centralises the {@code (hashCode & 0x7FFFFFFF) % partitions} striping so that components which must + * agree on a partition for the same key derive it identically. In particular, batched RPC persistence + * relies on this: {@code TbSqlBlockingQueueWrapper} picks the DB write partition and {@code TbRpcService} + * picks the post-persist callback stripe, both keyed off the rpcId hash. If those two ever disagreed on + * how the index is computed, the submission-order guarantee (e.g. RPC_QUEUED before RPC_DELIVERED) would + * silently break — keeping the math in one place removes that risk. + */ +public final class HashPartitioner { + + private HashPartitioner() { + } + + /** + * @param hashCode hash of the partition key + * @param partitions number of partitions; must be {@code > 0} + * @return a non-negative partition index in {@code [0, partitions)} + */ + public static int resolvePartition(int hashCode, int partitions) { + return (hashCode & 0x7FFFFFFF) % partitions; + } +} 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/RpcInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java index 9d0eb45464..08fa1c7969 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java @@ -26,6 +26,8 @@ import java.sql.PreparedStatement; import java.sql.SQLException; import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; @Repository public class RpcInsertRepository extends AbstractInsertRepository { @@ -50,8 +52,10 @@ public class RpcInsertRepository extends AbstractInsertRepository { } public List saveOrUpdate(List entries) { - List inserts = entries.stream().filter(RpcQueueEntry::insert).map(RpcQueueEntry::entity).toList(); - List updates = entries.stream().filter(entry -> !entry.insert()).map(RpcQueueEntry::entity).toList(); + Map> byIntent = entries.stream().collect(Collectors.partitioningBy( + RpcQueueEntry::insert, Collectors.mapping(RpcQueueEntry::entity, Collectors.toList()))); + List inserts = byIntent.get(true); + List updates = byIntent.get(false); int[] updateCounts = transactionTemplate.execute(status -> { // Inserts run first so a create and a status update for the same rpcId coalesced into one // batch still apply in create -> update order. From 9bd758d9c740b182b17ec1ab07035047770d92c2 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 19:24:08 +0300 Subject: [PATCH 08/12] refactor(rpc): harden batched upsert and partition config - build saveOrUpdate result inside the transaction so updateCounts is a non-null local taken straight from batchUpdate, avoiding an NPE on the @Nullable transactionTemplate.execute() result - guard HashPartitioner against a non-positive partition count - fail fast in TbRpcService when sql.rpc.callback_threads < 1 - add a deterministic saveOrUpdate coalesced-batch test and rename the timing-dependent async test to reflect what it actually verifies --- .../server/service/rpc/TbRpcService.java | 9 +-- .../common/util/HashPartitioner.java | 15 +--- .../server/dao/sql/rpc/JpaRpcDao.java | 3 - .../dao/sql/rpc/RpcInsertRepository.java | 41 ++++------- .../server/dao/sql/rpc/RpcQueueEntry.java | 5 -- .../server/dao/sql/rpc/JpaRpcDaoTest.java | 71 ++++++++++++------- 6 files changed, 64 insertions(+), 80 deletions(-) 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 e2f1e00fb1..e6d7f06c48 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 @@ -49,16 +49,13 @@ public class TbRpcService { private final RpcService rpcService; private final TbClusterService tbClusterService; - // Post-persist rule-engine notifications run on these striped single-thread executors, keyed by - // rpcId, instead of inline on the SQL persist threads. A flush batch can carry up to batch_size - // entries, and each notification's pushMsgToRuleEngine may do a (potentially blocking) device-profile - // cache lookup; running them inline would serialize that work on the persist thread and stall the - // next DB flush. Striping by rpcId (rather than a shared multi-threaded pool) also preserves - // per-command notification order, e.g. RPC_QUEUED before RPC_DELIVERED. 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]; 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 index 54555b2639..69efbd7ca4 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java +++ b/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java @@ -17,25 +17,16 @@ package org.thingsboard.common.util; /** * Maps a hash code to a non-negative partition index in {@code [0, partitions)}. - *

- * Centralises the {@code (hashCode & 0x7FFFFFFF) % partitions} striping so that components which must - * agree on a partition for the same key derive it identically. In particular, batched RPC persistence - * relies on this: {@code TbSqlBlockingQueueWrapper} picks the DB write partition and {@code TbRpcService} - * picks the post-persist callback stripe, both keyed off the rpcId hash. If those two ever disagreed on - * how the index is computed, the submission-order guarantee (e.g. RPC_QUEUED before RPC_DELIVERED) would - * silently break — keeping the math in one place removes that risk. */ public final class HashPartitioner { private HashPartitioner() { } - /** - * @param hashCode hash of the partition key - * @param partitions number of partitions; must be {@code > 0} - * @return a non-negative partition index in {@code [0, partitions)} - */ 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/sql/rpc/JpaRpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java index d7fe95b3ca..8752d58a78 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 @@ -71,9 +71,6 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; - // Boolean response per queued write: an INSERT-on-conflict always persists (true), while an - // UPDATE-by-id reports false when it matched no row (the RPC was deleted in the meantime), so the - // service layer can skip the rule-engine notification for a row that no longer exists. private TbSqlBlockingQueueWrapper queue; @PostConstruct diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java index 08fa1c7969..6b39f990ae 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java @@ -32,17 +32,11 @@ import java.util.stream.Collectors; @Repository public class RpcInsertRepository extends AbstractInsertRepository { - // Used only for the initial persistence of a new RPC (RpcStatus.QUEUED / already-EXPIRED on arrival). - // ON CONFLICT keeps the create idempotent (e.g. on actor re-processing); COALESCE never clobbers an - // existing response with NULL. private static final String INSERT = "INSERT INTO rpc (id, created_time, tenant_id, device_id, expiration_time, request, response, additional_info, status) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " + "ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, response = COALESCE(EXCLUDED.response, rpc.response);"; - // Used for every subsequent status change. WHERE id = ? means a row deleted in the meantime - // (TTL cleanup / manual delete) is NOT resurrected - it matches the old findById-null skip. - // COALESCE preserves a previously stored response when a status update carries none. private static final String UPDATE = "UPDATE rpc SET status = ?, response = COALESCE(?, response) WHERE id = ?;"; @@ -51,14 +45,12 @@ public class RpcInsertRepository extends AbstractInsertRepository { void bind(PreparedStatement ps, RpcEntity rpc) throws SQLException; } - public List saveOrUpdate(List entries) { + List saveOrUpdate(List entries) { Map> byIntent = entries.stream().collect(Collectors.partitioningBy( RpcQueueEntry::insert, Collectors.mapping(RpcQueueEntry::entity, Collectors.toList()))); List inserts = byIntent.get(true); List updates = byIntent.get(false); - int[] updateCounts = transactionTemplate.execute(status -> { - // Inserts run first so a create and a status update for the same rpcId coalesced into one - // batch still apply in create -> update order. + return transactionTemplate.execute(status -> { if (!inserts.isEmpty()) { batch(INSERT, inserts, (ps, rpc) -> { ps.setObject(1, rpc.getUuid()); @@ -72,26 +64,19 @@ public class RpcInsertRepository extends AbstractInsertRepository { ps.setString(9, rpc.getStatus().name()); }); } - if (!updates.isEmpty()) { - return batch(UPDATE, updates, (ps, rpc) -> { - ps.setString(1, rpc.getStatus().name()); - ps.setString(2, toJsonStr(rpc.getResponse())); - ps.setObject(3, rpc.getUuid()); - }); + int[] updateCounts = updates.isEmpty() ? new int[0] : batch(UPDATE, updates, (ps, rpc) -> { + ps.setString(1, rpc.getStatus().name()); + ps.setString(2, toJsonStr(rpc.getResponse())); + ps.setObject(3, rpc.getUuid()); + }); + + List persisted = new ArrayList<>(entries.size()); + int updateIdx = 0; + for (RpcQueueEntry entry : entries) { + persisted.add(entry.insert() || updateCounts[updateIdx++] > 0); } - return new int[0]; + return persisted; }); - - // Result is aligned to the submission order of entries: an insert always persists - // (INSERT ... ON CONFLICT), an update persists only if its WHERE id = ? matched a still-existing - // row. updateCounts keeps the same relative order as the filtered updates, so a single cursor - // walks it as we encounter update entries; insert entries short-circuit and don't advance it. - List persisted = new ArrayList<>(entries.size()); - int updateIdx = 0; - for (RpcQueueEntry entry : entries) { - persisted.add(entry.insert() || updateCounts[updateIdx++] > 0); - } - return persisted; } private int[] batch(String sql, List entities, ColumnBinder binder) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java index 8b5a5f96cf..a74fe7a2b4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java @@ -17,11 +17,6 @@ package org.thingsboard.server.dao.sql.rpc; import org.thingsboard.server.dao.model.sql.RpcEntity; -/** - * Pairs an {@link RpcEntity} with the write intent for the batched persistence queue, so the entity - * itself stays pure data. {@code insert} selects INSERT-on-conflict (initial create) vs UPDATE-by-id - * (status change that must never resurrect a deleted row). - */ record RpcQueueEntry(RpcEntity entity, boolean insert) { static RpcQueueEntry forInsert(RpcEntity entity) { 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 ad3f82c5bd..e2df9db08e 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 @@ -25,7 +25,9 @@ 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; @@ -36,6 +38,9 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { @Autowired JpaRpcDao rpcDao; + @Autowired + RpcInsertRepository rpcInsertRepository; + @Test public void deleteOutdated() { Rpc rpc = new Rpc(); @@ -83,34 +88,16 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { } @Test - public void saveAsyncSameRpcIdCoalescedBatchKeepsLatestStatus() throws Exception { + public void saveAsyncSequentialWritesConvergeToLatestStatus() throws Exception { UUID id = UUID.randomUUID(); DeviceId deviceId = new DeviceId(UUID.randomUUID()); - long createdTime = System.currentTimeMillis(); - long expirationTime = createdTime + 60_000; - - Rpc queued = new Rpc(new RpcId(id)); - queued.setCreatedTime(createdTime); - queued.setTenantId(TenantId.SYS_TENANT_ID); - queued.setDeviceId(deviceId); - queued.setExpirationTime(expirationTime); - queued.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}")); - queued.setStatus(RpcStatus.QUEUED); - - Rpc delivered = new Rpc(new RpcId(id)); - delivered.setCreatedTime(createdTime); - delivered.setTenantId(TenantId.SYS_TENANT_ID); - delivered.setDeviceId(deviceId); - delivered.setExpirationTime(expirationTime); - delivered.setRequest(queued.getRequest()); - delivered.setStatus(RpcStatus.DELIVERED); - delivered.setResponse(JacksonUtil.toJsonNode("{\"ok\":true}")); - - // Enqueue both writes for the same rpcId back-to-back so they coalesce into one flush batch. - // Same rpcId -> same partition; the queue's stable sort must keep submission order - // (QUEUED before DELIVERED), so the final persisted row must be DELIVERED, never QUEUED. - var queuedFuture = rpcDao.createAsync(queued); - var deliveredFuture = rpcDao.updateAsync(delivered); + + // A create followed by a status update for the same rpcId, both via the async API. Whether the + // two land in one flush batch or two is up to the queue's timing and not asserted here (see + // saveOrUpdateCoalescedBatchAppliesInOrderAndAlignsResults for the deterministic coalesced case); + // either way the final persisted row must converge to DELIVERED, never get stuck at QUEUED. + var queuedFuture = rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)); + var deliveredFuture = rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.DELIVERED, JacksonUtil.toJsonNode("{\"ok\":true}"))); queuedFuture.get(5, TimeUnit.SECONDS); deliveredFuture.get(5, TimeUnit.SECONDS); @@ -120,6 +107,38 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { assertThat(stored.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); } + @Test + public void saveOrUpdateCoalescedBatchAppliesInOrderAndAlignsResults() { + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + UUID idA = UUID.randomUUID(); + UUID idB = UUID.randomUUID(); // never inserted -> its update must report no match + + // Drive the persist logic directly with a single, deterministically-coalesced flush batch. + // This is exactly what "coalescing" means: one saveOrUpdate call carrying several writes for + // the same partition in submission order. No queue timing involved. + // index 0: create A (QUEUED) -> INSERT, always persists -> true + // index 1: update B (SUCCESSFUL) -> UPDATE for a missing row -> false + // index 2: update A (DELIVERED, {ok:true}) -> UPDATE on the row inserted at index 0 -> true + List batch = List.of( + RpcQueueEntry.forInsert(new RpcEntity(rpc(idA, deviceId, RpcStatus.QUEUED, null))), + RpcQueueEntry.forUpdate(new RpcEntity(rpc(idB, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"x\":1}")))), + RpcQueueEntry.forUpdate(new RpcEntity(rpc(idA, deviceId, RpcStatus.DELIVERED, JacksonUtil.toJsonNode("{\"ok\":true}"))))); + + List persisted = rpcInsertRepository.saveOrUpdate(batch); + + // Booleans are aligned positionally to submission order even though saveOrUpdate runs all + // inserts before all updates internally - this guards the updateIdx cursor alignment. + assertThat(persisted).containsExactly(true, false, true); + + // Inserts run before updates, so A ends DELIVERED (never stuck at QUEUED)... + Rpc storedA = rpcDao.findById(TenantId.SYS_TENANT_ID, idA); + assertThat(storedA).isNotNull(); + assertThat(storedA.getStatus()).isEqualTo(RpcStatus.DELIVERED); + 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(); From 2757172ef333df755cf8ce199bfa2c3e7438c086 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Mon, 29 Jun 2026 12:40:35 +0300 Subject: [PATCH 09/12] fix(rpc): preserve original createdTime in post-persist RE notifications On the update paths buildRpc rebuilt the Rpc with createdTime=now, which the post-persist rule-engine notification then serialized, so RPC_DELIVERED/etc. carried the update moment instead of the row's real creation time (the UPDATE never touches created_time). Capture createdTime once at create and thread it through ToDeviceRpcRequestMetadata so update-path notifications report the original value, restoring pre-async behavior. Also add a striped-executor ordering test (same rpcId -> RPC_QUEUED before RPC_DELIVERED) and document that sql.rpc.callback_threads is independent of batch_threads. --- .../device/DeviceActorMessageProcessor.java | 31 +++++++------- .../device/ToDeviceRpcRequestMetadata.java | 3 ++ .../src/main/resources/thingsboard.yml | 2 +- .../server/service/rpc/TbRpcServiceTest.java | 40 +++++++++++++++++-- 4 files changed, 57 insertions(+), 19 deletions(-) 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 6d0f07e507..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,13 +258,13 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso }; } - private void createRpc(ToDeviceRpcRequest request, RpcStatus status) { - systemContext.getTbRpcService().create(tenantId, buildRpc(request, status, null)); + 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) { + 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()); @@ -339,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()); } @@ -356,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().update(tenantId, buildRpc(toDeviceRpcRequest, 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)); @@ -667,7 +668,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } catch (IllegalArgumentException e) { response = JacksonUtil.newObjectNode().put("error", payload); } - systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRequestMsg, status, response)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRequestMsg, status, response, requestMd.getCreatedTime())); } } finally { if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE)) { @@ -731,7 +732,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } if (persisted) { - systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, 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) { @@ -824,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().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout, md.getCreatedTime())); } }, systemContext.getRpcResponseTimeout(), TimeUnit.MILLISECONDS); } @@ -1037,7 +1038,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso rpc.setStatus(RpcStatus.EXPIRED); 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/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index af1b7d328d..5cda04a33d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -429,7 +429,7 @@ sql: 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. 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 index cf11dfed48..7190e1cb69 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -18,18 +18,22 @@ 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 static org.junit.Assert.assertEquals; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; @@ -88,11 +92,41 @@ public class TbRpcServiceTest { .pushMsgToRuleEngine(any(TenantId.class), any(DeviceId.class), any(TbMsg.class), isNull()); } + @Test + public void sameRpcIdNotificationsRunInSubmissionOrderOnStripedExecutor() { + // Use several stripes so executorFor actually partitions. Both writes share an rpcId, so they + // must map to the same stripe and the rule-engine notifications must arrive in submission order + // (RPC_QUEUED before RPC_DELIVERED) - this is the ordering guarantee executorFor exists for. + tbRpcService = new TbRpcService(rpcService, clusterService, 3); + + RpcId rpcId = new RpcId(UUID.randomUUID()); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + Rpc queued = newRpc(rpcId, deviceId, RpcStatus.QUEUED); + Rpc delivered = newRpc(rpcId, deviceId, RpcStatus.DELIVERED); + when(rpcService.createAsync(queued)).thenReturn(Futures.immediateFuture(true)); + when(rpcService.updateAsync(delivered)).thenReturn(Futures.immediateFuture(true)); + + tbRpcService.create(queued.getTenantId(), queued); + tbRpcService.update(delivered.getTenantId(), delivered); + + 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_QUEUED, msgs.get(0).getInternalType()); + assertEquals(TbMsgType.RPC_DELIVERED, msgs.get(1).getInternalType()); + } + private Rpc newRpc() { - Rpc rpc = new Rpc(new RpcId(UUID.randomUUID())); + 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(new DeviceId(UUID.randomUUID())); - rpc.setStatus(RpcStatus.QUEUED); + rpc.setDeviceId(deviceId); + rpc.setStatus(status); rpc.setRequest(JacksonUtil.toJsonNode("{}")); return rpc; } From a5b496ba22efeb71a6bcc9c232e3145ad830501a Mon Sep 17 00:00:00 2001 From: dshvaika Date: Mon, 29 Jun 2026 13:12:57 +0300 Subject: [PATCH 10/12] test(rpc): make striped-ordering test actually guard per-rpcId serialization The previous version stubbed both writes with immediateFuture and submitted them sequentially, so the QUEUED-before-DELIVERED order held by construction and would have passed even if per-rpcId striping regressed to a shared multi-threaded pool. Make the QUEUED notification block (signal start via latch, then sleep) and submit DELIVERED only once QUEUED is in flight, so the assertion now depends on the single-thread stripe serializing same-rpcId callbacks. --- .../server/service/rpc/TbRpcServiceTest.java | 27 ++++++++++++++++--- 1 file changed, 23 insertions(+), 4 deletions(-) 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 index 7190e1cb69..300a9ed8d3 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -32,12 +32,16 @@ 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.timeout; import static org.mockito.Mockito.verify; @@ -93,10 +97,9 @@ public class TbRpcServiceTest { } @Test - public void sameRpcIdNotificationsRunInSubmissionOrderOnStripedExecutor() { - // Use several stripes so executorFor actually partitions. Both writes share an rpcId, so they - // must map to the same stripe and the rule-engine notifications must arrive in submission order - // (RPC_QUEUED before RPC_DELIVERED) - this is the ordering guarantee executorFor exists for. + public void sameRpcIdNotificationsRunInSubmissionOrderOnStripedExecutor() throws InterruptedException { + // The count is incidental here: a single rpcId always maps to one stripe. What the test really + // checks is that two notifications for the SAME rpcId are serialized on that one stripe. tbRpcService = new TbRpcService(rpcService, clusterService, 3); RpcId rpcId = new RpcId(UUID.randomUUID()); @@ -106,7 +109,23 @@ public class TbRpcServiceTest { when(rpcService.createAsync(queued)).thenReturn(Futures.immediateFuture(true)); when(rpcService.updateAsync(delivered)).thenReturn(Futures.immediateFuture(true)); + // Make the QUEUED notification block while it runs: it signals that it has started, then sleeps - + // holding the stripe. If the two callbacks for this rpcId were NOT serialized on one stripe, the + // fast DELIVERED callback would overtake the sleeping QUEUED one and be recorded first. + CountDownLatch queuedStarted = new CountDownLatch(1); + doAnswer(invocation -> { + TbMsg msg = invocation.getArgument(2); + if (msg.getInternalType() == TbMsgType.RPC_QUEUED) { + queuedStarted.countDown(); + Thread.sleep(300); + } + return null; + }).when(clusterService).pushMsgToRuleEngine(eq(TenantId.SYS_TENANT_ID), eq(deviceId), any(TbMsg.class), isNull()); + tbRpcService.create(queued.getTenantId(), queued); + // Don't submit DELIVERED until QUEUED is actually in flight (and now sleeping) on the stripe - + // this makes the test about stripe serialization, not about submission timing. + assertTrue(queuedStarted.await(5, TimeUnit.SECONDS)); tbRpcService.update(delivered.getTenantId(), delivered); ArgumentCaptor msgCaptor = ArgumentCaptor.forClass(TbMsg.class); From 7e35f7b184ad20ddb572d1b9b4fff03cf1f01835 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Tue, 30 Jun 2026 16:25:44 +0300 Subject: [PATCH 11/12] fix(rpc): persist QUEUED RPC synchronously before sending to device --- .../thingsboard/server/service/rpc/TbRpcService.java | 12 +++++++++++- .../server/service/rpc/TbRpcServiceTest.java | 9 ++++++--- 2 files changed, 17 insertions(+), 4 deletions(-) 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 e6d7f06c48..0e11658451 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 @@ -73,7 +73,8 @@ public class TbRpcService { } public void create(TenantId tenantId, Rpc rpc) { - persist(tenantId, rpc, rpcService.createAsync(rpc)); + rpcService.save(rpc); + executorFor(rpc.getUuidId()).execute(() -> notifyRuleEngine(tenantId, rpc)); } public void update(TenantId tenantId, Rpc rpc) { @@ -99,6 +100,15 @@ public class TbRpcService { 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); + } + } + private void pushRpcMsgToRuleEngine(TenantId tenantId, Rpc rpc) { TbMsg msg = TbMsg.newMsg() .type(TbMsgType.valueOf("RPC_" + rpc.getStatus().name())) 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 index 300a9ed8d3..585da09b43 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -43,6 +43,7 @@ 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; @@ -71,13 +72,15 @@ public class TbRpcServiceTest { } @Test - public void createPersistsViaCreateAsyncThenPushesToRuleEngine() { + public void createPersistsSynchronouslyThenPushesToRuleEngine() { Rpc rpc = newRpc(); - when(rpcService.createAsync(rpc)).thenReturn(Futures.immediateFuture(true)); + when(rpcService.save(rpc)).thenReturn(rpc); tbRpcService.create(rpc.getTenantId(), rpc); - verify(rpcService).createAsync(rpc); + // create must persist synchronously via save(...), never via the async batch path + verify(rpcService).save(rpc); + verify(rpcService, never()).createAsync(any()); verify(clusterService, timeout(5000)) .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); } From 24905f1f9ad1183ea31e4deb4c4ee1b35f6e953f Mon Sep 17 00:00:00 2001 From: dshvaika Date: Tue, 30 Jun 2026 16:29:46 +0300 Subject: [PATCH 12/12] test(rpc): re-aim striped-ordering test at two status updates --- .../server/service/rpc/TbRpcServiceTest.java | 33 +++++++++---------- 1 file changed, 16 insertions(+), 17 deletions(-) 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 index 585da09b43..513b667c3c 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -100,44 +100,43 @@ public class TbRpcServiceTest { } @Test - public void sameRpcIdNotificationsRunInSubmissionOrderOnStripedExecutor() throws InterruptedException { - // The count is incidental here: a single rpcId always maps to one stripe. What the test really - // checks is that two notifications for the SAME rpcId are serialized on that one stripe. + 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 queued = newRpc(rpcId, deviceId, RpcStatus.QUEUED); Rpc delivered = newRpc(rpcId, deviceId, RpcStatus.DELIVERED); - when(rpcService.createAsync(queued)).thenReturn(Futures.immediateFuture(true)); + 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 QUEUED notification block while it runs: it signals that it has started, then sleeps - - // holding the stripe. If the two callbacks for this rpcId were NOT serialized on one stripe, the - // fast DELIVERED callback would overtake the sleeping QUEUED one and be recorded first. - CountDownLatch queuedStarted = new CountDownLatch(1); + // 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_QUEUED) { - queuedStarted.countDown(); + 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.create(queued.getTenantId(), queued); - // Don't submit DELIVERED until QUEUED is actually in flight (and now sleeping) on the stripe - - // this makes the test about stripe serialization, not about submission timing. - assertTrue(queuedStarted.await(5, TimeUnit.SECONDS)); 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_QUEUED, msgs.get(0).getInternalType()); - assertEquals(TbMsgType.RPC_DELIVERED, msgs.get(1).getInternalType()); + assertEquals(TbMsgType.RPC_DELIVERED, msgs.get(0).getInternalType()); + assertEquals(TbMsgType.RPC_SUCCESSFUL, msgs.get(1).getInternalType()); } private Rpc newRpc() {