From 25d3394e9f0700abad5f0250fb5c1704a86c3829 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Thu, 26 Mar 2026 11:10:39 +0100 Subject: [PATCH 1/4] Fix race condition in notification deduplication check The alreadyProcessed() method used separate get() and put() calls on the local cache, allowing concurrent threads in the notification executor pool to both read null and bypass deduplication, creating duplicate notifications. Replace with a single compute() call that atomically checks and updates the cache entry, preventing the race between concurrent trigger processing. Also fix: discard external cache timestamps that are more than 1 hour in the future (clock skew protection), and avoid reading back from the SOFT ref local cache when writing to external cache (GC could null it out). --- ...faultNotificationDeduplicationService.java | 56 +++--- ...tNotificationDeduplicationServiceTest.java | 160 ++++++++++++++++++ 2 files changed, 192 insertions(+), 24 deletions(-) create mode 100644 common/queue/src/test/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationServiceTest.java diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationService.java b/common/queue/src/main/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationService.java index bf884958eb..279a2ddc7f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationService.java @@ -32,6 +32,7 @@ import org.thingsboard.server.queue.util.PropertyUtils; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; import static org.springframework.util.ConcurrentReferenceHashMap.ReferenceType.SOFT; @@ -59,41 +60,48 @@ public class DefaultNotificationDeduplicationService implements NotificationDedu } private boolean alreadyProcessed(NotificationRuleTrigger trigger, String deduplicationKey, boolean onlyLocalCache) { - Long lastProcessedTs = localCache.get(deduplicationKey); - if (lastProcessedTs == null && !onlyLocalCache) { - Cache externalCache = getExternalCache(); - if (externalCache != null) { - lastProcessedTs = externalCache.get(deduplicationKey, Long.class); - } else { - log.warn("Sent notifications cache is not set up"); + long deduplicationDuration = getDeduplicationDuration(trigger); + final long now = System.currentTimeMillis(); + boolean[] result = {false}; + + localCache.compute(deduplicationKey, (key, lastProcessedTs) -> { + if (lastProcessedTs == null && !onlyLocalCache) { + Cache externalCache = getExternalCache(); + if (externalCache != null) { + lastProcessedTs = externalCache.get(key, Long.class); + if (lastProcessedTs != null && lastProcessedTs > now + TimeUnit.HOURS.toMillis(1)) { + log.warn("Discarding dedup entry from external cache for key '{}': timestamp is {} ms in the future", + key, lastProcessedTs - now); + lastProcessedTs = null; + } + } else { + log.warn("Sent notifications cache is not set up"); + } } - } - boolean alreadyProcessed = false; - long deduplicationDuration = getDeduplicationDuration(trigger); - if (lastProcessedTs != null) { - long passed = System.currentTimeMillis() - lastProcessedTs; - log.trace("Deduplicating trigger {} by key '{}'. Deduplication duration: {} ms, passed: {} ms", - trigger.getType(), deduplicationKey, deduplicationDuration, passed); - if (deduplicationDuration == 0 || passed <= deduplicationDuration) { - alreadyProcessed = true; + if (lastProcessedTs != null) { + long passed = now - lastProcessedTs; + log.trace("Deduplicating trigger {} by key '{}'. Deduplication duration: {} ms, passed: {} ms", + trigger.getType(), key, deduplicationDuration, passed); + if (deduplicationDuration == 0 || passed <= deduplicationDuration) { + result[0] = true; + return lastProcessedTs; + } } - } - if (!alreadyProcessed) { - lastProcessedTs = System.currentTimeMillis(); - } - localCache.put(deduplicationKey, lastProcessedTs); + return now; + }); + if (!onlyLocalCache) { - if (!alreadyProcessed || deduplicationDuration == 0) { + if (!result[0] || deduplicationDuration == 0) { // if lastProcessedTs is changed or if deduplicating infinitely (so that cache value not removed by ttl) Cache externalCache = getExternalCache(); if (externalCache != null) { - externalCache.put(deduplicationKey, lastProcessedTs); + externalCache.put(deduplicationKey, now); } } } - return alreadyProcessed; + return result[0]; } public static String getDeduplicationKey(NotificationRuleTrigger trigger, NotificationRule rule) { diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationServiceTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationServiceTest.java new file mode 100644 index 0000000000..826c1e1ef6 --- /dev/null +++ b/common/queue/src/test/java/org/thingsboard/server/queue/notification/DefaultNotificationDeduplicationServiceTest.java @@ -0,0 +1,160 @@ +/** + * 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.queue.notification; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.cache.Cache; +import org.springframework.cache.CacheManager; +import org.springframework.cache.concurrent.ConcurrentMapCacheManager; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.common.data.CacheConstants; +import org.thingsboard.server.common.data.notification.rule.NotificationRule; +import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class DefaultNotificationDeduplicationServiceTest { + + private static final int TIMEOUT = 30; + + private DefaultNotificationDeduplicationService deduplicationService; + private CacheManager cacheManager; + + @BeforeEach + void setUp() { + deduplicationService = new DefaultNotificationDeduplicationService(); + deduplicationService.setDeduplicationDurations(""); + cacheManager = new ConcurrentMapCacheManager(CacheConstants.SENT_NOTIFICATIONS_CACHE); + ReflectionTestUtils.setField(deduplicationService, "cacheManager", cacheManager); + } + + @Test + void testFirstTriggerIsNotDeduplicated() { + NotificationRuleTrigger trigger = mockTrigger(TimeUnit.HOURS.toMillis(1)); + NotificationRule rule = mockRule(); + + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isFalse(); + } + + @Test + void testSecondTriggerIsDeduplicated() { + NotificationRuleTrigger trigger = mockTrigger(TimeUnit.HOURS.toMillis(1)); + NotificationRule rule = mockRule(); + + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isFalse(); + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isTrue(); + } + + @Test + void testTriggerPassesAfterDeduplicationWindowExpires() { + NotificationRuleTrigger trigger = mockTrigger(50); // 50ms dedup window + NotificationRule rule = mockRule(); + + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isFalse(); + + try { + Thread.sleep(200); // wait well past the 50ms window + } catch (InterruptedException ignored) {} + + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isFalse(); + } + + @Test + void testFutureTimestampFromExternalCacheIsDiscarded() { + NotificationRuleTrigger trigger = mockTrigger(TimeUnit.HOURS.toMillis(1)); + NotificationRule rule = mockRule(); + String dedupKey = DefaultNotificationDeduplicationService.getDeduplicationKey(trigger, rule); + + // Put a timestamp 2 hours in the future into external cache + Cache externalCache = cacheManager.getCache(CacheConstants.SENT_NOTIFICATIONS_CACHE); + externalCache.put(dedupKey, System.currentTimeMillis() + TimeUnit.HOURS.toMillis(2)); + + // Should NOT be deduplicated — future timestamp must be discarded + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isFalse(); + } + + @Test + void testValidTimestampFromExternalCacheIsDeduplicated() { + NotificationRuleTrigger trigger = mockTrigger(TimeUnit.HOURS.toMillis(1)); + NotificationRule rule = mockRule(); + String dedupKey = DefaultNotificationDeduplicationService.getDeduplicationKey(trigger, rule); + + // Put a recent timestamp into external cache + Cache externalCache = cacheManager.getCache(CacheConstants.SENT_NOTIFICATIONS_CACHE); + externalCache.put(dedupKey, System.currentTimeMillis()); + + // Should be deduplicated — valid external cache entry + assertThat(deduplicationService.alreadyProcessed(trigger, rule)).isTrue(); + } + + @Test + void testConcurrentTriggersProduceExactlyOneNonDeduplicated() throws Exception { + NotificationRuleTrigger trigger = mockTrigger(TimeUnit.HOURS.toMillis(1)); + NotificationRule rule = mockRule(); + + int threadCount = 10; + CyclicBarrier barrier = new CyclicBarrier(threadCount); + List results = new CopyOnWriteArrayList<>(); + + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + try { + for (int i = 0; i < threadCount; i++) { + executor.submit(() -> { + try { + barrier.await(TIMEOUT, TimeUnit.SECONDS); + } catch (Exception ignored) {} + results.add(deduplicationService.alreadyProcessed(trigger, rule)); + }); + } + executor.shutdown(); + assertThat(executor.awaitTermination(TIMEOUT, TimeUnit.SECONDS)).isTrue(); + + assertThat(results).hasSize(threadCount); + assertThat(results.stream().filter(r -> !r).count()) + .as("exactly one trigger should pass through deduplication") + .isEqualTo(1); + } finally { + executor.shutdownNow(); + } + } + + private NotificationRuleTrigger mockTrigger(long deduplicationDurationMs) { + NotificationRuleTrigger trigger = mock(NotificationRuleTrigger.class); + when(trigger.getType()).thenReturn(NotificationRuleTriggerType.RESOURCES_SHORTAGE); + when(trigger.getDeduplicationKey()).thenReturn("test:dedup:key"); + when(trigger.getDefaultDeduplicationDuration()).thenReturn(deduplicationDurationMs); + when(trigger.getDeduplicationStrategy()).thenReturn(NotificationRuleTrigger.DeduplicationStrategy.ONLY_MATCHING); + return trigger; + } + + private NotificationRule mockRule() { + NotificationRule rule = mock(NotificationRule.class); + when(rule.getDeduplicationKey()).thenReturn("rule:key"); + return rule; + } + +} From a75c71eeb6d9de454af6758ddb9ad6b0159f1dd0 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Mon, 27 Apr 2026 15:00:56 +0300 Subject: [PATCH 2/4] added spring compression properties --- application/src/main/resources/thingsboard.yml | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index e1195b6d3b..43a2a23207 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -58,6 +58,14 @@ server: http2: # Enable/disable HTTP/2 support enabled: "${HTTP2_ENABLED:true}" + # HTTP response compression + compression: + # Enable/disable HTTP response compression + enabled: "${SERVER_COMPRESSION_ENABLED:false}" + # Minimum size (in bytes) required for a response before compression is applied + min-response-size: "${SERVER_COMPRESSION_MIN_RESPONSE_SIZE:2048}" + # Comma-separated list of MIME types that should be compressed + mime-types: "${SERVER_COMPRESSION_MIME_TYPES:text/html,text/xml,text/plain,text/css,text/javascript,application/javascript,application/json,application/xml}" # Log errors with stacktrace when REST API throws an exception with the message "Please contact sysadmin" log_controller_error_stack_trace: "${HTTP_LOG_CONTROLLER_ERROR_STACK_TRACE:false}" ws: From ebfc12038daccb73505230bfa278d7f04e82344f Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Wed, 29 Apr 2026 16:56:47 +0300 Subject: [PATCH 3/4] env renaming --- application/src/main/resources/thingsboard.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 43a2a23207..cced6915be 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -61,11 +61,11 @@ server: # HTTP response compression compression: # Enable/disable HTTP response compression - enabled: "${SERVER_COMPRESSION_ENABLED:false}" + enabled: "${HTTP_COMPRESSION_ENABLED:false}" # Minimum size (in bytes) required for a response before compression is applied - min-response-size: "${SERVER_COMPRESSION_MIN_RESPONSE_SIZE:2048}" + min_response_size: "${HTTP_COMPRESSION_MIN_RESPONSE_SIZE:2048}" # Comma-separated list of MIME types that should be compressed - mime-types: "${SERVER_COMPRESSION_MIME_TYPES:text/html,text/xml,text/plain,text/css,text/javascript,application/javascript,application/json,application/xml}" + mime_types: "${HTTP_COMPRESSION_MIME_TYPES:text/html,text/xml,text/plain,text/css,text/javascript,application/javascript,application/json,application/xml}" # Log errors with stacktrace when REST API throws an exception with the message "Please contact sysadmin" log_controller_error_stack_trace: "${HTTP_LOG_CONTROLLER_ERROR_STACK_TRACE:false}" ws: From 4be45d447520aad7f261899a37462556baad4495 Mon Sep 17 00:00:00 2001 From: Maksym Tsymbarov Date: Wed, 29 Apr 2026 17:07:22 +0200 Subject: [PATCH 4/4] Fixed CVE-2026-40895 --- msa/web-ui/yarn.lock | 6 +++--- ui-ngx/yarn.lock | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/msa/web-ui/yarn.lock b/msa/web-ui/yarn.lock index f421a62684..033aeac686 100644 --- a/msa/web-ui/yarn.lock +++ b/msa/web-ui/yarn.lock @@ -774,9 +774,9 @@ fn.name@1.x.x: integrity sha512-GRnmB5gPyJpAhTQdSZTSp9uaPSvl09KoYcMQtsB9rQoOmzs9dH6ffeccH+Z+cv6P68Hu5bC6JjRh4Ah/mHSNRw== follow-redirects@^1.0.0: - version "1.15.11" - resolved "https://registry.yarnpkg.com/follow-redirects/-/follow-redirects-1.15.11.tgz#777d73d72a92f8ec4d2e410eb47352a56b8e8340" - integrity sha512-deG2P0JfjrTxl50XGCDyfI97ZGVCxIpfKYmfyrQ54n5FO/0gfIES8C/Psl6kWVDolizcaaxZJnTS0QSMxvnsBQ== + version "1.16.0" + resolved "https://registry.yarnpkg.com/follow-redirects/-/follow-redirects-1.16.0.tgz#28474a159d3b9d11ef62050a14ed60e4df6d61bc" + integrity sha512-y5rN/uOsadFT/JfYwhxRS5R7Qce+g3zG97+JrtFZlC9klX/W5hD7iiLzScI4nZqUS7DNUdhPgw4xI8W2LuXlUw== forwarded@0.2.0: version "0.2.0" diff --git a/ui-ngx/yarn.lock b/ui-ngx/yarn.lock index 4da832c885..e9564f5279 100644 --- a/ui-ngx/yarn.lock +++ b/ui-ngx/yarn.lock @@ -6240,9 +6240,9 @@ flatted@^3.2.9: resolved "https://github.com/thingsboard/flot.git#c2734540477d8b261d04ee18d4d38af3b0ecb81b" follow-redirects@^1.0.0: - version "1.15.9" - resolved "https://registry.yarnpkg.com/follow-redirects/-/follow-redirects-1.15.9.tgz#a604fa10e443bf98ca94228d9eebcc2e8a2c8ee1" - integrity sha512-gew4GsXizNgdoRyqmyfMHyAmXsZDk6mHkSxZFCzW9gwlbtOW44CDtYavM+y+72qD/Vq2l550kMF52DT8fOLJqQ== + version "1.16.0" + resolved "https://registry.yarnpkg.com/follow-redirects/-/follow-redirects-1.16.0.tgz#28474a159d3b9d11ef62050a14ed60e4df6d61bc" + integrity sha512-y5rN/uOsadFT/JfYwhxRS5R7Qce+g3zG97+JrtFZlC9klX/W5hD7iiLzScI4nZqUS7DNUdhPgw4xI8W2LuXlUw== font-awesome@^4.7.0: version "4.7.0"