From c01302fe6bd349f5a8524ce8e9ec3133a4d7f50f Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Tue, 21 Oct 2025 18:20:45 +0300 Subject: [PATCH] added lock on entity creation to fix race condition on multiple entity creation --- .../server/dao/asset/BaseAssetService.java | 4 ++ .../dao/customer/CustomerServiceImpl.java | 2 +- .../dao/dashboard/DashboardServiceImpl.java | 4 ++ .../server/dao/device/DeviceServiceImpl.java | 4 ++ .../server/dao/edge/EdgeServiceImpl.java | 4 ++ .../dao/entity/AbstractEntityService.java | 23 ++++++++ .../server/dao/rule/BaseRuleChainService.java | 4 ++ .../server/dao/user/UserServiceImpl.java | 4 ++ .../dao/service/AbstractServiceTest.java | 6 ++ .../server/dao/service/AssetServiceTest.java | 57 +++++++++++++++++++ .../server/dao/service/DeviceServiceTest.java | 31 ++++++++++ 11 files changed, 142 insertions(+), 1 deletion(-) diff --git a/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java b/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java index fed201d403..3e49c6d6b3 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java @@ -148,6 +148,10 @@ public class BaseAssetService extends AbstractCachedEntityService doSaveAsset(asset, doValidate)); + } + + private Asset doSaveAsset(Asset asset, boolean doValidate) { log.trace("Executing saveAsset [{}]", asset); Asset oldAsset = null; if (doValidate) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/customer/CustomerServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/customer/CustomerServiceImpl.java index b06902d95e..fa1de490fa 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/customer/CustomerServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/customer/CustomerServiceImpl.java @@ -139,7 +139,7 @@ public class CustomerServiceImpl extends AbstractCachedEntityService saveCustomer(customer, true)); } private Customer saveCustomer(Customer customer, boolean doValidate) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/dashboard/DashboardServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/dashboard/DashboardServiceImpl.java index 1b21310237..b044f8fbe7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/dashboard/DashboardServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/dashboard/DashboardServiceImpl.java @@ -157,6 +157,10 @@ public class DashboardServiceImpl extends AbstractEntityService implements Dashb @Override public Dashboard saveDashboard(Dashboard dashboard, boolean doValidate) { + return saveLimitedEntity(dashboard, () -> doSaveDashboard(dashboard, doValidate)); + } + + private Dashboard doSaveDashboard(Dashboard dashboard, boolean doValidate) { log.trace("Executing saveDashboard [{}]", dashboard); if (doValidate) { dashboardValidator.validate(dashboard, DashboardInfo::getTenantId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java index 6d993f3e3d..cd64ddccda 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java @@ -210,6 +210,10 @@ public class DeviceServiceImpl extends CachedVersionedEntityService doSaveDeviceWithoutCredentials(device, doValidate)); + } + + private Device doSaveDeviceWithoutCredentials(Device device, boolean doValidate) { log.trace("Executing saveDevice [{}]", device); Device oldDevice = null; if (doValidate) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java index 0655d05572..b71decf5f6 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java @@ -201,6 +201,10 @@ public class EdgeServiceImpl extends AbstractCachedEntityService doSaveEdge(edge)); + } + + private Edge doSaveEdge(Edge edge) { log.trace("Executing saveEdge [{}]", edge); Edge oldEdge = edgeValidator.validate(edge, Edge::getTenantId); EdgeCacheEvictEvent evictEvent = new EdgeCacheEvictEvent(edge.getTenantId(), edge.getName(), oldEdge != null ? oldEdge.getName() : null); diff --git a/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java b/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java index 7560c7fb76..8a8f74671b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java @@ -21,13 +21,16 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Lazy; +import org.springframework.util.ConcurrentReferenceHashMap; import org.thingsboard.common.util.DebugModeUtil; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.HasDebugSettings; +import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.debug.DebugSettings; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.HasId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; @@ -44,7 +47,10 @@ import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; @Slf4j public abstract class AbstractEntityService { @@ -52,6 +58,8 @@ public abstract class AbstractEntityService { public static final String INCORRECT_EDGE_ID = "Incorrect edgeId "; public static final String INCORRECT_PAGE_LINK = "Incorrect page link "; + private final ConcurrentMap entityCreationLocks = new ConcurrentReferenceHashMap<>(16); + @Autowired protected ApplicationEventPublisher eventPublisher; @@ -86,6 +94,21 @@ public abstract class AbstractEntityService { @Value("${debug.settings.default_duration:15}") private int defaultDebugDurationMinutes; + protected E saveLimitedEntity(E entity, Supplier saveFunction) { + log.debug("Creating limited entity: {}", entity); + if (entity.getId() == null) { + ReentrantLock lock = entityCreationLocks.computeIfAbsent(entity.getTenantId(), id -> new ReentrantLock()); + lock.lock(); + try { + return saveFunction.get(); + } finally { + lock.unlock(); + } + } else { + return saveFunction.get(); + } + } + protected void createRelation(TenantId tenantId, EntityRelation relation) { log.debug("Creating relation: {}", relation); relationService.saveRelation(tenantId, relation); diff --git a/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java b/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java index 8538bd9492..47e82f7df1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java @@ -125,6 +125,10 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC @Override @Transactional public RuleChain saveRuleChain(RuleChain ruleChain, boolean publishSaveEvent, boolean doValidate) { + return saveLimitedEntity(ruleChain, () -> doSaveRuleChain(ruleChain, publishSaveEvent, true)); + } + + private RuleChain doSaveRuleChain(RuleChain ruleChain, boolean publishSaveEvent, boolean doValidate) { log.trace("Executing doSaveRuleChain [{}]", ruleChain); if (doValidate) { ruleChainValidator.validate(ruleChain, RuleChain::getTenantId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java index 5c94ba1891..8b92c5f8ac 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java @@ -159,6 +159,10 @@ public class UserServiceImpl extends AbstractCachedEntityService doSaveUser(tenantId, user)); + } + + private User doSaveUser(TenantId tenantId, User user) { log.trace("Executing saveUser [{}]", user); User oldUser = userValidator.validate(user, User::getTenantId); if (!userLoginCaseSensitive) { diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java index 25339244fd..0d2b9b5d9f 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java @@ -48,6 +48,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.HasId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.id.TenantProfileId; import org.thingsboard.server.common.data.oauth2.MapperType; import org.thingsboard.server.common.data.oauth2.OAuth2Client; import org.thingsboard.server.common.data.oauth2.OAuth2CustomMapperConfig; @@ -185,8 +186,13 @@ public abstract class AbstractServiceTest { } public Tenant createTenant() { + return createTenant(null); + } + + public Tenant createTenant(TenantProfileId tenantProfileId) { Tenant tenant = new Tenant(); tenant.setTitle("My tenant " + UUID.randomUUID()); + tenant.setTenantProfileId(tenantProfileId); Tenant savedTenant = tenantService.saveTenant(tenant); assertNotNull(savedTenant); return savedTenant; diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/AssetServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/AssetServiceTest.java index 462e7a894c..ccd0a1ca9b 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/AssetServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/AssetServiceTest.java @@ -16,17 +16,25 @@ package org.thingsboard.server.dao.service; import com.datastax.oss.driver.api.core.uuid.Uuids; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; +import org.junit.After; import org.junit.Assert; +import org.junit.Before; import org.junit.Test; import org.junit.jupiter.api.Assertions; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.DefaultTransactionDefinition; +import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; +import org.testcontainers.shaded.org.awaitility.Awaitility; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.asset.AssetProfile; @@ -44,6 +52,8 @@ import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; +import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; import org.thingsboard.server.dao.asset.AssetDao; import org.thingsboard.server.dao.asset.AssetProfileService; import org.thingsboard.server.dao.asset.AssetService; @@ -51,11 +61,14 @@ import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.relation.RelationService; +import org.thingsboard.server.dao.tenant.TenantProfileService; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; @@ -72,6 +85,8 @@ public class AssetServiceTest extends AbstractServiceTest { @Autowired RelationService relationService; @Autowired + TenantProfileService tenantProfileService; + @Autowired private AssetProfileService assetProfileService; @Autowired private CalculatedFieldService calculatedFieldService; @@ -79,6 +94,18 @@ public class AssetServiceTest extends AbstractServiceTest { private PlatformTransactionManager platformTransactionManager; private IdComparator idComparator = new IdComparator<>(); + ListeningExecutorService executor; + private TenantId anotherTenantId; + + @Before + public void before() { + executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10, ThingsBoardThreadFactory.forName(getClass().getSimpleName() + "-test-scope"))); + } + + @After + public void after() { + executor.shutdownNow(); + } @Test public void testSaveAsset() { @@ -105,6 +132,36 @@ public class AssetServiceTest extends AbstractServiceTest { assetService.deleteAsset(tenantId, savedAsset.getId()); } + @Test + public void testAssetLimitOnTenantProfileLevel() { + TenantProfile tenantProfile = new TenantProfile(); + tenantProfile.setName("Test profile"); + tenantProfile.setDescription("Test"); + TenantProfileData profileData = new TenantProfileData(); + profileData.setConfiguration(DefaultTenantProfileConfiguration.builder().maxAssets(5l).build()); + tenantProfile.setProfileData(profileData); + tenantProfile.setDefault(false); + tenantProfile.setIsolatedTbRuleEngine(false); + + tenantProfile = tenantProfileService.saveTenantProfile(anotherTenantId, tenantProfile); + anotherTenantId = createTenant(tenantProfile.getId()).getId(); + + for (int i = 0; i < 20; i++) { + executor.submit(() -> { + Asset asset = new Asset(); + asset.setTenantId(anotherTenantId); + asset.setName(RandomStringUtils.randomAlphabetic(10)); + asset.setType("default"); + assetService.saveAsset(asset); + }); + } + + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> { + long countByTenantId = assetService.countByTenantId(anotherTenantId); + return countByTenantId == 5; + }); + } + @Test public void testShouldNotPutInCacheRolledbackAssetProfile() { AssetProfile assetProfile = new AssetProfile(); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java index 32767043d7..257eed8232 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java @@ -16,6 +16,8 @@ package org.thingsboard.server.dao.service; import com.datastax.oss.driver.api.core.uuid.Uuids; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -27,6 +29,8 @@ import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.DefaultTransactionDefinition; +import org.testcontainers.shaded.org.awaitility.Awaitility; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceInfo; @@ -74,6 +78,8 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; @@ -105,10 +111,12 @@ public class DeviceServiceTest extends AbstractServiceTest { private IdComparator idComparator = new IdComparator<>(); private TenantId anotherTenantId; + private ListeningExecutorService executor; @Before public void before() { anotherTenantId = createTenant().getId(); + executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10, ThingsBoardThreadFactory.forName(getClass().getSimpleName() + "-test-scope"))); } @After @@ -118,6 +126,7 @@ public class DeviceServiceTest extends AbstractServiceTest { tenantProfileService.deleteTenantProfiles(tenantId); tenantProfileService.deleteTenantProfiles(anotherTenantId); + executor.shutdownNow(); } @Test @@ -136,6 +145,28 @@ public class DeviceServiceTest extends AbstractServiceTest { deleteDevice(tenantId, device); } + @Test + public void testDeviceLimitOnTenantProfileLevel() { + TenantProfile defaultTenantProfile = tenantProfileService.findDefaultTenantProfile(tenantId); + defaultTenantProfile.getProfileData().setConfiguration(DefaultTenantProfileConfiguration.builder().maxDevices(5l).build()); + tenantProfileService.saveTenantProfile(tenantId, defaultTenantProfile); + + for (int i = 0; i < 20; i++) { + executor.submit(() -> { + Device device = new Device(); + device.setTenantId(tenantId); + device.setName(StringUtils.randomAlphabetic(10)); + device.setType("default"); + deviceService.saveDevice(device, true); + }); + } + + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> { + long countByTenantId = deviceService.countByTenantId(tenantId); + return countByTenantId == 5; + }); + } + @Test public void testSaveDevicesWithMaxDeviceOutOfLimit() { TenantProfile defaultTenantProfile = tenantProfileService.findDefaultTenantProfile(tenantId);