From 361e267db4aeddc9b791b69e48fe2798bed4f670 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Fri, 5 Jun 2026 15:06:55 +0300 Subject: [PATCH 1/9] Edge: prevent disconnect notification spam from flapping edges --- .../service/edge/rpc/EdgeGrpcService.java | 104 ++++++++++++++++-- .../src/main/resources/thingsboard.yml | 5 + .../edge/EdgeConnectionNotificationTest.java | 96 ++++++++++++++++ 3 files changed, 194 insertions(+), 11 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 072a4878d5..7f57101205 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -41,7 +41,6 @@ import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; -import org.thingsboard.server.common.transport.config.ssl.PemSslCredentials; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.EdgeId; @@ -58,6 +57,7 @@ import org.thingsboard.server.common.msg.edge.EdgeHighPriorityMsg; import org.thingsboard.server.common.msg.edge.EdgeSessionMsg; import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse; import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest; +import org.thingsboard.server.common.transport.config.ssl.PemSslCredentials; import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; import org.thingsboard.server.gen.edge.v1.RequestMsg; import org.thingsboard.server.gen.edge.v1.ResponseMsg; @@ -68,7 +68,6 @@ import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.EdgeContextComponent; -import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import java.io.IOException; import java.util.ArrayList; @@ -103,6 +102,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private final ConcurrentMap sessionNewEventsLocks = new ConcurrentHashMap<>(); private final Map sessionNewEvents = new HashMap<>(); private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>(); + private final ConcurrentMap> pendingDisconnectNotifications = new ConcurrentHashMap<>(); private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>(); private final ConcurrentMap edgeEventsMigrationProcessed = new ConcurrentHashMap<>(); private final List zombieSessions = new ArrayList<>(); @@ -136,6 +136,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @Value("${edges.max_high_priority_queue_size_per_session:10000}") private int maxHighPriorityQueueSizePerSession; + @Value("${edges.connectivity.disconnect_notification_delay_ms:60000}") + private long disconnectNotificationDelayMs; + @Autowired @Lazy private EdgeContextComponent ctx; @@ -239,6 +242,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i sessionEdgeEventChecks.remove(edgeId); } } + pendingDisconnectNotifications.values().forEach(task -> { + if (task != null && !task.isDone()) { + task.cancel(false); + } + }); + pendingDisconnectNotifications.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); } @@ -321,6 +330,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } + cancelPendingDisconnectNotification(edgeId); cancelScheduleEdgeEventsCheck(edgeId); } } @@ -383,7 +393,11 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i long lastConnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_CONNECT_TIME, lastConnectTs); edgeIdServiceIdCache.put(edgeId, serviceInfoProvider.getServiceId()); - pushRuleEngineMessage(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); + // If the edge reconnected within the disconnect-notification delay window, suppress the pending + // "disconnected" notification - the drop was transient (debounce for flapping edges). + cancelPendingDisconnectNotification(edgeId); + pushStateEventToRuleEngine(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); + notifyEdgeConnectivity(tenantId, edge, true); cancelScheduleEdgeEventsCheck(edgeId); edgeEventsMigrationProcessed.putIfAbsent(edgeId, Boolean.FALSE); scheduleEdgeEventsCheck(edgeGrpcSession); @@ -545,7 +559,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i save(tenantId, edgeId, ACTIVITY_STATE, false); long lastDisconnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_DISCONNECT_TIME, lastDisconnectTs); - pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); + pushStateEventToRuleEngine(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); + scheduleDisconnectNotification(tenantId, edge); cancelScheduleEdgeEventsCheck(edgeId); } else { log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId); @@ -617,7 +632,33 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - private void pushRuleEngineMessage(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) { + private static class AttributeSaveCallback implements FutureCallback { + + private final TenantId tenantId; + private final EdgeId edgeId; + private final String key; + private final Object value; + + AttributeSaveCallback(TenantId tenantId, EdgeId edgeId, String key, Object value) { + this.tenantId = tenantId; + this.edgeId = edgeId; + this.key = key; + this.value = value; + } + + @Override + public void onSuccess(@Nullable Void result) { + log.trace("[{}][{}] Successfully updated attribute [{}] with value [{}]", tenantId, edgeId, key, value); + } + + @Override + public void onFailure(Throwable t) { + log.warn("[{}][{}] Failed to update attribute [{}] with value [{}]", tenantId, edgeId, key, value, t); + } + + } + + private void pushStateEventToRuleEngine(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) { try { EdgeId edgeId = edge.getId(); ObjectNode edgeState = JacksonUtil.newObjectNode(); @@ -629,12 +670,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i edgeState.put(ACTIVITY_STATE, false); edgeState.put(LAST_DISCONNECT_TIME, ts); } - ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() - .tenantId(tenantId) - .customerId(edge.getCustomerId()) - .edgeId(edgeId) - .edgeName(edge.getName()) - .connected(isConnected).build()); String data = JacksonUtil.toString(edgeState); TbMsgMetaData md = new TbMsgMetaData(); if (!persistToTelemetry) { @@ -655,6 +690,53 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } + private void notifyEdgeConnectivity(TenantId tenantId, Edge edge, boolean connected) { + try { + ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() + .tenantId(tenantId) + .customerId(edge.getCustomerId()) + .edgeId(edge.getId()) + .edgeName(edge.getName()) + .connected(connected).build()); + } catch (Exception e) { + log.warn("[{}][{}] Failed to process edge connectivity notification (connected={})", tenantId, edge.getId(), connected, e); + } + } + + private void scheduleDisconnectNotification(TenantId tenantId, Edge edge) { + if (disconnectNotificationDelayMs <= 0) { + notifyEdgeConnectivity(tenantId, edge, false); + return; + } + EdgeId edgeId = edge.getId(); + pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { + if (existing != null && !existing.isDone()) { + existing.cancel(false); + } + return executorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), + disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); + }); + } + + private void fireDelayedDisconnectNotification(TenantId tenantId, Edge edge) { + EdgeId edgeId = edge.getId(); + pendingDisconnectNotifications.remove(edgeId); + // Re-verify the edge is still disconnected. The cache is cluster-wide, so this also covers the case + // where the edge dropped on this node and reconnected to a different TB-Core node within the window. + if (sessions.containsKey(edgeId) || edgeIdServiceIdCache.get(edgeId) != null) { + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", tenantId, edgeId); + return; + } + notifyEdgeConnectivity(tenantId, edge, false); + } + + private void cancelPendingDisconnectNotification(EdgeId edgeId) { + ScheduledFuture pending = pendingDisconnectNotifications.remove(edgeId); + if (pending != null && !pending.isDone()) { + pending.cancel(false); + } + } + private void cleanupZombieSessions() { try { tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId())); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 454c59d7d7..0cb3e57ae8 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1642,6 +1642,11 @@ edges: state: # Persist state of edge (active, last connect, last disconnect) into timeseries or attributes tables. 'false' means to store edge state into attributes table persistToTelemetry: "${EDGES_PERSIST_STATE_TO_TELEMETRY:false}" + connectivity: + # Delay (ms) before sending an edge "disconnected" notification. Suppressed if the edge reconnects within + # this window - debounces flapping edges from spamming the notification center. Only the notification is + # delayed; rule-engine events and state attributes update immediately. Set to 0 to notify immediately. + disconnect_notification_delay_ms: "${TB_EDGE_DISCONNECT_NOTIFICATION_DELAY_MS:60000}" stats: # Enable or disable reporting of edge communication stats (true or false) enabled: "${EDGES_STATS_ENABLED:true}" diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java new file mode 100644 index 0000000000..f72ad51f4a --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -0,0 +1,96 @@ +/** + * 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.edge; + +import org.junit.Test; +import org.mockito.ArgumentMatcher; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; +import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; +import org.thingsboard.server.controller.AbstractWebTest; +import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.edge.imitator.EdgeImitator; +import org.thingsboard.server.gen.edge.v1.OAuth2ClientUpdateMsg; +import org.thingsboard.server.gen.edge.v1.OAuth2DomainUpdateMsg; + +import java.util.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@DaoSqlTest +@TestPropertySource(properties = { + "edges.connectivity.disconnect_notification_delay_ms=5000" +}) +public class EdgeConnectionNotificationTest extends AbstractEdgeTest { + + private static final long DELAY_MS = 5000L; + + @MockitoSpyBean + private NotificationRuleProcessor notificationRuleProcessor; + + @Test + public void givenEdgeStaysDisconnected_whenDelayElapses_thenDisconnectNotificationSent() throws Exception { + clearInvocations(notificationRuleProcessor); + + edgeImitator.disconnect(); + + // After the configured delay, the "disconnected" notification is sent exactly once. + await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> + verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger(false)))); + } + + @Test + public void givenEdgeReconnectsWithinDelay_whenEdgeFlaps_thenDisconnectNotificationSuppressed() throws Exception { + clearInvocations(notificationRuleProcessor); + + // Edge drops... + edgeImitator.disconnect(); + // Ensure the server processed the disconnect (and scheduled the delayed notification) before reconnecting. + TimeUnit.SECONDS.sleep(1); + + // ...and reconnects within the delay window, which must cancel the pending "disconnected" notification. + EdgeImitator reconnected = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); + reconnected.ignoreType(OAuth2ClientUpdateMsg.class); + reconnected.ignoreType(OAuth2DomainUpdateMsg.class); + reconnected.connect(); + edgeImitator = reconnected; // let teardown clean up the live session + + // The "connected" notification still fires immediately on reconnect (we suppress the disconnect only). + await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> + verify(notificationRuleProcessor, atLeastOnce()).process(argThat(edgeConnectionTrigger(true)))); + + // Wait until the original disconnect-notification window has fully elapsed... + TimeUnit.MILLISECONDS.sleep(DELAY_MS + 1000); + + // ...the "disconnected" notification must have never been sent. + verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false))); + } + + private ArgumentMatcher edgeConnectionTrigger(boolean connected) { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger + && edge.getId().equals(edgeTrigger.getEdgeId()) + && edgeTrigger.isConnected() == connected; + } + +} From 4993ffc66f4749d2465730b498b9f3ea5b5a4131 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Mon, 8 Jun 2026 17:07:28 +0300 Subject: [PATCH 2/9] refactor: address PR #15732 review - cancel helper, pooled executor, test coverage --- .../service/edge/rpc/EdgeGrpcService.java | 36 +++++++---- .../src/main/resources/thingsboard.yml | 2 +- .../server/edge/AbstractEdgeTest.java | 14 +++- .../edge/EdgeConnectionNotificationTest.java | 6 +- ...geImmediateDisconnectNotificationTest.java | 64 +++++++++++++++++++ 5 files changed, 99 insertions(+), 23 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 7f57101205..a1910396dc 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -36,6 +36,7 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; +import org.thingsboard.server.cache.TbCacheValueWrapper; import org.thingsboard.server.cache.TbTransactionalCache; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; @@ -242,11 +243,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i sessionEdgeEventChecks.remove(edgeId); } } - pendingDisconnectNotifications.values().forEach(task -> { - if (task != null && !task.isDone()) { - task.cancel(false); - } - }); + pendingDisconnectNotifications.values().forEach(EdgeGrpcService::cancelIfPending); pendingDisconnectNotifications.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); @@ -330,9 +327,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } - cancelPendingDisconnectNotification(edgeId); cancelScheduleEdgeEventsCheck(edgeId); } + cancelPendingDisconnectNotification(edgeId); } private void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) { @@ -576,7 +573,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId); } } - edgeIdServiceIdCache.evict(edgeId); + evictServiceIdCacheIfOwnedByThisNode(edgeId); + } + + // Only evict if the cache still points to this node. If the edge already reconnected to a different + // TB-Core node within the keep-alive window, that node has overwritten the entry - evicting it here + // would wipe the live owner and make fireDelayedDisconnectNotification raise a false 'disconnected'. + private void evictServiceIdCacheIfOwnedByThisNode(EdgeId edgeId) { + TbCacheValueWrapper wrapper = edgeIdServiceIdCache.get(edgeId); + if (wrapper != null && serviceInfoProvider.getServiceId().equals(wrapper.get())) { + edgeIdServiceIdCache.evict(edgeId); + } } private void destroySession(EdgeGrpcSession session) { @@ -710,10 +717,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } EdgeId edgeId = edge.getId(); pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { - if (existing != null && !existing.isDone()) { - existing.cancel(false); - } - return executorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), + cancelIfPending(existing); + return edgeEventProcessingExecutorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); }); } @@ -731,9 +736,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void cancelPendingDisconnectNotification(EdgeId edgeId) { - ScheduledFuture pending = pendingDisconnectNotifications.remove(edgeId); - if (pending != null && !pending.isDone()) { - pending.cancel(false); + cancelIfPending(pendingDisconnectNotifications.remove(edgeId)); + } + + private static void cancelIfPending(ScheduledFuture future) { + if (future != null && !future.isDone()) { + future.cancel(false); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 0cb3e57ae8..07ea4dbf06 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1646,7 +1646,7 @@ edges: # Delay (ms) before sending an edge "disconnected" notification. Suppressed if the edge reconnects within # this window - debounces flapping edges from spamming the notification center. Only the notification is # delayed; rule-engine events and state attributes update immediately. Set to 0 to notify immediately. - disconnect_notification_delay_ms: "${TB_EDGE_DISCONNECT_NOTIFICATION_DELAY_MS:60000}" + disconnect_notification_delay_ms: "${EDGES_DISCONNECT_NOTIFICATION_DELAY_MS:60000}" stats: # Enable or disable reporting of edge communication stats (true or false) enabled: "${EDGES_STATS_ENABLED:true}" diff --git a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java index d75c7b8f59..71802468ab 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -125,6 +125,7 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { public static final String EDGE_HOST = "localhost"; public static final int EDGE_PORT = TestSocketUtils.findAvailableTcpPort(); + @DynamicPropertySource static void props(DynamicPropertyRegistry registry) { log.debug("edges.rpc.port = {}", EDGE_PORT); @@ -157,16 +158,23 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { //8 installation messages installation(); - edgeImitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); + edgeImitator = createEdgeImitator(); // 17 connect messages + 8 installation messages edgeImitator.expectMessageAmount(SYNC_MESSAGE_COUNT); - edgeImitator.ignoreType(OAuth2ClientUpdateMsg.class); - edgeImitator.ignoreType(OAuth2DomainUpdateMsg.class); edgeImitator.connect(); verifyEdgeConnectionAndInitialData(); } + // Creates an EdgeImitator wired with the standard ignored message types, but not yet connected. + // Callers add any expectations (e.g. expectMessageAmount) before invoking connect() themselves. + protected EdgeImitator createEdgeImitator() throws Exception { + EdgeImitator imitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); + imitator.ignoreType(OAuth2ClientUpdateMsg.class); + imitator.ignoreType(OAuth2DomainUpdateMsg.class); + return imitator; + } + @After public void teardownEdgeTest() { try { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java index f72ad51f4a..8179e5894a 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -25,8 +25,6 @@ import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.controller.AbstractWebTest; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.edge.imitator.EdgeImitator; -import org.thingsboard.server.gen.edge.v1.OAuth2ClientUpdateMsg; -import org.thingsboard.server.gen.edge.v1.OAuth2DomainUpdateMsg; import java.util.concurrent.TimeUnit; @@ -70,9 +68,7 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { TimeUnit.SECONDS.sleep(1); // ...and reconnects within the delay window, which must cancel the pending "disconnected" notification. - EdgeImitator reconnected = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); - reconnected.ignoreType(OAuth2ClientUpdateMsg.class); - reconnected.ignoreType(OAuth2DomainUpdateMsg.class); + EdgeImitator reconnected = createEdgeImitator(); reconnected.connect(); edgeImitator = reconnected; // let teardown clean up the live session diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java new file mode 100644 index 0000000000..a9d6dbe455 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java @@ -0,0 +1,64 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.edge; + +import org.junit.Test; +import org.mockito.ArgumentMatcher; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; +import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; +import org.thingsboard.server.controller.AbstractWebTest; +import org.thingsboard.server.dao.service.DaoSqlTest; + +import java.util.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +// Covers the disconnect_notification_delay_ms <= 0 branch in EdgeGrpcService.scheduleDisconnectNotification, +// where the "disconnected" notification must be sent immediately instead of being scheduled with a delay. +@DaoSqlTest +@TestPropertySource(properties = { + "edges.connectivity.disconnect_notification_delay_ms=0" +}) +public class EdgeImmediateDisconnectNotificationTest extends AbstractEdgeTest { + + @MockitoSpyBean + private NotificationRuleProcessor notificationRuleProcessor; + + @Test + public void givenZeroDelay_whenEdgeDisconnects_thenDisconnectNotificationSentImmediately() throws Exception { + clearInvocations(notificationRuleProcessor); + + edgeImitator.disconnect(); + + // With a zero delay there is no debounce window - the "disconnected" notification fires right away. + await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> + verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger()))); + } + + private ArgumentMatcher edgeConnectionTrigger() { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger + && edge.getId().equals(edgeTrigger.getEdgeId()) + && !edgeTrigger.isConnected(); + } + +} From f10ca1338c99c22148bad5a026435e916bb69be3 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 10 Jun 2026 16:34:02 +0300 Subject: [PATCH 3/9] removeOnCancelPolicy, cluster-path unit test, cleanups --- .../service/edge/rpc/EdgeGrpcService.java | 90 +++++--- .../server/edge/AbstractEdgeTest.java | 1 - .../edge/EdgeConnectionNotificationTest.java | 56 ++++- ...geImmediateDisconnectNotificationTest.java | 64 ------ .../service/edge/rpc/EdgeGrpcServiceTest.java | 204 ++++++++++++++++++ .../common/util/ThingsBoardExecutors.java | 10 +- 6 files changed, 317 insertions(+), 108 deletions(-) delete mode 100644 application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java create mode 100644 application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index a1910396dc..b875226034 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -69,6 +69,7 @@ import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.EdgeContextComponent; +import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import java.io.IOException; import java.util.ArrayList; @@ -103,7 +104,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private final ConcurrentMap sessionNewEventsLocks = new ConcurrentHashMap<>(); private final Map sessionNewEvents = new HashMap<>(); private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>(); - private final ConcurrentMap> pendingDisconnectNotifications = new ConcurrentHashMap<>(); + private final ConcurrentMap pendingDisconnectNotifications = new ConcurrentHashMap<>(); private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>(); private final ConcurrentMap edgeEventsMigrationProcessed = new ConcurrentHashMap<>(); private final List zombieSessions = new ArrayList<>(); @@ -144,6 +145,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @Lazy private EdgeContextComponent ctx; + @Autowired + private TelemetrySubscriptionService tsSubService; + @Autowired private TbClusterService clusterService; @@ -198,7 +202,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } this.edgeEventProcessingExecutorService = ThingsBoardExecutors.newScheduledThreadPool(schedulerPoolSize, "edge-event-check-scheduler"); this.sendDownlinkExecutorService = ThingsBoardExecutors.newScheduledThreadPool(sendSchedulerPoolSize, "edge-send-scheduler"); - this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service"); + // removeOnCancelPolicy: cancelled delayed disconnect notifications (common with flapping edges) are dropped + // from the queue right away on reconnect, instead of piling up until their original fire time. + this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service", true); this.executorService.scheduleAtFixedRate(this::cleanupZombieSessions, 60, 60, TimeUnit.SECONDS); log.info("Edge RPC service initialized!"); } @@ -232,19 +238,23 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @PreDestroy public void destroy() { + // Flush already-pending disconnect notifications BEFORE shutting the server down. These are edges that + // disconnected on their own within the delay window; without this a graceful restart would silently + // swallow their "disconnected" alert. Snapshotting first means we only flush those - not edges that this + // very shutdown is about to disconnect (they rebalance to another node and shouldn't alert) - and we run + // while the scheduler and rule engine are still alive. The guard inside fireDelayedDisconnectNotification + // still suppresses any edge that reconnected elsewhere. + List pendingToFlush = new ArrayList<>(pendingDisconnectNotifications.values()); + pendingDisconnectNotifications.clear(); + pendingToFlush.forEach(pending -> { + cancelIfPending(pending.future()); + fireDelayedDisconnectNotification(pending); + }); if (server != null) { server.shutdownNow(); } - for (Map.Entry> entry : sessionEdgeEventChecks.entrySet()) { - EdgeId edgeId = entry.getKey(); - ScheduledFuture sessionEdgeEventCheck = entry.getValue(); - if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) { - sessionEdgeEventCheck.cancel(true); - sessionEdgeEventChecks.remove(edgeId); - } - } - pendingDisconnectNotifications.values().forEach(EdgeGrpcService::cancelIfPending); - pendingDisconnectNotifications.clear(); + sessionEdgeEventChecks.values().forEach(task -> cancelIfPending(task, true)); + sessionEdgeEventChecks.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); } @@ -393,6 +403,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i // If the edge reconnected within the disconnect-notification delay window, suppress the pending // "disconnected" notification - the drop was transient (debounce for flapping edges). cancelPendingDisconnectNotification(edgeId); + // Connect notifies immediately; only the disconnect notification is debounced (see scheduleDisconnectNotification). pushStateEventToRuleEngine(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); notifyEdgeConnectivity(tenantId, edge, true); cancelScheduleEdgeEventsCheck(edgeId); @@ -528,13 +539,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void cancelScheduleEdgeEventsCheck(EdgeId edgeId) { log.trace("[{}] cancelling edge event check for edge", edgeId); - if (sessionEdgeEventChecks.containsKey(edgeId)) { - ScheduledFuture sessionEdgeEventCheck = sessionEdgeEventChecks.get(edgeId); - if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) { - sessionEdgeEventCheck.cancel(true); - sessionEdgeEventChecks.remove(edgeId); - } - } + cancelIfPending(sessionEdgeEventChecks.remove(edgeId), true); } private void onEdgeDisconnect(Edge edge, UUID sessionId) { @@ -602,14 +607,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void save(TenantId tenantId, EdgeId edgeId, String key, long value) { log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { - ctx.getTsSubService().saveTimeseries(TimeseriesSaveRequest.builder() + tsSubService.saveTimeseries(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .entry(new LongDataEntry(key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .build()); } else { - ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder() + tsSubService.saveAttributes(AttributesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .scope(AttributeScope.SERVER_SCOPE) @@ -622,14 +627,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void save(TenantId tenantId, EdgeId edgeId, String key, boolean value) { log.debug("[{}][{}] Updating boolean edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { - ctx.getTsSubService().saveTimeseries(TimeseriesSaveRequest.builder() + tsSubService.saveTimeseries(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .entry(new BooleanDataEntry(key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .build()); } else { - ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder() + tsSubService.saveAttributes(AttributesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .scope(AttributeScope.SERVER_SCOPE) @@ -717,31 +722,51 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } EdgeId edgeId = edge.getId(); pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { - cancelIfPending(existing); - return edgeEventProcessingExecutorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), + if (existing != null) { + cancelIfPending(existing.future()); + } + // Single-element holder so the scheduled task can reference its own pending entry, which doesn't exist + // until schedule(...) returns. The task needs it for the identity-keyed remove in fireDelayedDisconnectNotification. + PendingDisconnect[] holder = {null}; + ScheduledFuture future = executorService.schedule( + () -> fireDelayedDisconnectNotification(holder[0]), disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); + holder[0] = new PendingDisconnect(tenantId, edge, future); + return holder[0]; }); } - private void fireDelayedDisconnectNotification(TenantId tenantId, Edge edge) { - EdgeId edgeId = edge.getId(); - pendingDisconnectNotifications.remove(edgeId); + private void fireDelayedDisconnectNotification(PendingDisconnect pending) { + EdgeId edgeId = pending.edge().getId(); + // Identity-keyed remove: don't clobber a newer entry if a second disconnect races with this task firing. + pendingDisconnectNotifications.remove(edgeId, pending); // Re-verify the edge is still disconnected. The cache is cluster-wide, so this also covers the case // where the edge dropped on this node and reconnected to a different TB-Core node within the window. if (sessions.containsKey(edgeId) || edgeIdServiceIdCache.get(edgeId) != null) { - log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", tenantId, edgeId); + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", pending.tenantId(), edgeId); return; } - notifyEdgeConnectivity(tenantId, edge, false); + notifyEdgeConnectivity(pending.tenantId(), pending.edge(), false); } private void cancelPendingDisconnectNotification(EdgeId edgeId) { - cancelIfPending(pendingDisconnectNotifications.remove(edgeId)); + PendingDisconnect pending = pendingDisconnectNotifications.remove(edgeId); + if (pending != null) { + cancelIfPending(pending.future()); + } } + // Carries the context the delayed task needs, so a pending notification can still be fired on shutdown + // (see destroy()), not just cancelled. Package-private for unit testing of fireDelayedDisconnectNotification. + record PendingDisconnect(TenantId tenantId, Edge edge, ScheduledFuture future) {} + private static void cancelIfPending(ScheduledFuture future) { + cancelIfPending(future, false); + } + + private static void cancelIfPending(ScheduledFuture future, boolean mayInterruptIfRunning) { if (future != null && !future.isDone()) { - future.cancel(false); + future.cancel(mayInterruptIfRunning); } } @@ -797,6 +822,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i !kafkaSession.getConsumer().getConsumer().isStopped(); } return false; + } } diff --git a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java index 71802468ab..8a3c3c1ae7 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -125,7 +125,6 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { public static final String EDGE_HOST = "localhost"; public static final int EDGE_PORT = TestSocketUtils.findAvailableTcpPort(); - @DynamicPropertySource static void props(DynamicPropertyRegistry registry) { log.debug("edges.rpc.port = {}", EDGE_PORT); diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java index 8179e5894a..c46d82437a 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -15,16 +15,20 @@ */ package org.thingsboard.server.edge; +import org.junit.After; +import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentMatcher; -import org.springframework.test.context.TestPropertySource; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.springframework.test.util.ReflectionTestUtils; import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.controller.AbstractWebTest; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.edge.imitator.EdgeImitator; +import org.thingsboard.server.service.edge.rpc.EdgeGrpcService; import java.util.concurrent.TimeUnit; @@ -37,9 +41,6 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @DaoSqlTest -@TestPropertySource(properties = { - "edges.connectivity.disconnect_notification_delay_ms=5000" -}) public class EdgeConnectionNotificationTest extends AbstractEdgeTest { private static final long DELAY_MS = 5000L; @@ -47,19 +48,55 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { @MockitoSpyBean private NotificationRuleProcessor notificationRuleProcessor; + @Autowired + private EdgeGrpcService edgeGrpcService; + + private long originalDisconnectNotificationDelayMs; + + // Capture the bean's configured delay before each test and restore it after, so a test method that forgets + // to call setDisconnectNotificationDelayMs can't silently inherit the previous method's mutated value + // (the shared context means the mutation would otherwise persist across methods). + @Before + public void captureDisconnectNotificationDelay() { + originalDisconnectNotificationDelayMs = (long) ReflectionTestUtils.getField(edgeGrpcService, "disconnectNotificationDelayMs"); + } + + @After + public void restoreDisconnectNotificationDelay() { + setDisconnectNotificationDelayMs(originalDisconnectNotificationDelayMs); + } + + // The delay is overridden per test (rather than via a per-class @TestPropertySource) so all cases share a + // single Spring application context instead of booting a separate heavy context per delay value. + private void setDisconnectNotificationDelayMs(long delayMs) { + ReflectionTestUtils.setField(edgeGrpcService, "disconnectNotificationDelayMs", delayMs); + } + @Test public void givenEdgeStaysDisconnected_whenDelayElapses_thenDisconnectNotificationSent() throws Exception { + // After the configured delay, the "disconnected" notification is sent exactly once. + assertDisconnectNotificationSentOnce(DELAY_MS); + } + + @Test + public void givenZeroDelay_whenEdgeDisconnects_thenDisconnectNotificationSentImmediately() throws Exception { + // With a zero delay there is no debounce window - the "disconnected" notification fires right away. + assertDisconnectNotificationSentOnce(0); + } + + private void assertDisconnectNotificationSentOnce(long delayMs) throws Exception { + setDisconnectNotificationDelayMs(delayMs); clearInvocations(notificationRuleProcessor); edgeImitator.disconnect(); - // After the configured delay, the "disconnected" notification is sent exactly once. await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger(false)))); } @Test public void givenEdgeReconnectsWithinDelay_whenEdgeFlaps_thenDisconnectNotificationSuppressed() throws Exception { + setDisconnectNotificationDelayMs(DELAY_MS); clearInvocations(notificationRuleProcessor); // Edge drops... @@ -76,11 +113,10 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> verify(notificationRuleProcessor, atLeastOnce()).process(argThat(edgeConnectionTrigger(true)))); - // Wait until the original disconnect-notification window has fully elapsed... - TimeUnit.MILLISECONDS.sleep(DELAY_MS + 1000); - - // ...the "disconnected" notification must have never been sent. - verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false))); + // The "disconnected" notification must never be sent throughout the full delay window. + await().during(DELAY_MS + 500, TimeUnit.MILLISECONDS) + .atMost(DELAY_MS + 2000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false)))); } private ArgumentMatcher edgeConnectionTrigger(boolean connected) { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java deleted file mode 100644 index a9d6dbe455..0000000000 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java +++ /dev/null @@ -1,64 +0,0 @@ -/** - * 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.edge; - -import org.junit.Test; -import org.mockito.ArgumentMatcher; -import org.springframework.test.context.TestPropertySource; -import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; -import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; -import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; -import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; -import org.thingsboard.server.controller.AbstractWebTest; -import org.thingsboard.server.dao.service.DaoSqlTest; - -import java.util.concurrent.TimeUnit; - -import static org.awaitility.Awaitility.await; -import static org.mockito.ArgumentMatchers.argThat; -import static org.mockito.Mockito.clearInvocations; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; - -// Covers the disconnect_notification_delay_ms <= 0 branch in EdgeGrpcService.scheduleDisconnectNotification, -// where the "disconnected" notification must be sent immediately instead of being scheduled with a delay. -@DaoSqlTest -@TestPropertySource(properties = { - "edges.connectivity.disconnect_notification_delay_ms=0" -}) -public class EdgeImmediateDisconnectNotificationTest extends AbstractEdgeTest { - - @MockitoSpyBean - private NotificationRuleProcessor notificationRuleProcessor; - - @Test - public void givenZeroDelay_whenEdgeDisconnects_thenDisconnectNotificationSentImmediately() throws Exception { - clearInvocations(notificationRuleProcessor); - - edgeImitator.disconnect(); - - // With a zero delay there is no debounce window - the "disconnected" notification fires right away. - await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> - verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger()))); - } - - private ArgumentMatcher edgeConnectionTrigger() { - return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger - && edge.getId().equals(edgeTrigger.getEdgeId()) - && !edgeTrigger.isConnected(); - } - -} diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java new file mode 100644 index 0000000000..f538bd1baa --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java @@ -0,0 +1,204 @@ +/** + * 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.edge.rpc; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentMatcher; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.cache.SimpleTbCacheValueWrapper; +import org.thingsboard.server.cache.TbTransactionalCache; +import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; +import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.service.edge.EdgeContextComponent; + +import java.util.UUID; +import java.util.concurrent.ConcurrentMap; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Unit coverage for the cluster-aware predicates that guard the delayed disconnect notification: + * {@code evictServiceIdCacheIfOwnedByThisNode} and the re-verify check in + * {@code fireDelayedDisconnectNotification}. These exercise the cross-node ownership logic that the + * single-node integration tests ({@code EdgeConnectionNotificationTest}) cannot reach. + */ +@ExtendWith(MockitoExtension.class) +public class EdgeGrpcServiceTest { + + private static final String THIS_NODE = "tb-core-1"; + private static final String OTHER_NODE = "tb-core-2"; + + @Mock + private TbTransactionalCache edgeIdServiceIdCache; + + @Mock + private TbServiceInfoProvider serviceInfoProvider; + + @Mock + private EdgeContextComponent ctx; + + @Mock + private NotificationRuleProcessor ruleProcessor; + + @InjectMocks + private EdgeGrpcService edgeGrpcService; + + private TenantId tenantId; + private EdgeId edgeId; + private Edge edge; + + @BeforeEach + public void setUp() { + tenantId = new TenantId(UUID.randomUUID()); + edgeId = new EdgeId(UUID.randomUUID()); + edge = new Edge(edgeId); + edge.setTenantId(tenantId); + edge.setName("test-edge"); + } + + // --- evictServiceIdCacheIfOwnedByThisNode --- + + @Test + public void givenCacheOwnedByThisNode_whenEvict_thenEntryIsEvicted() { + when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(THIS_NODE)); + + evictServiceIdCacheIfOwnedByThisNode(); + + verify(edgeIdServiceIdCache, times(1)).evict(edgeId); + } + + @Test + public void givenCacheOwnedByAnotherNode_whenEvict_thenEntryIsKept() { + // The edge already reconnected to another node within the keep-alive window: must NOT wipe the live owner. + when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); + + evictServiceIdCacheIfOwnedByThisNode(); + + verify(edgeIdServiceIdCache, never()).evict(edgeId); + } + + @Test + public void givenEmptyCache_whenEvict_thenNothingEvicted() { + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); + + evictServiceIdCacheIfOwnedByThisNode(); + + verify(edgeIdServiceIdCache, never()).evict(edgeId); + } + + // --- fireDelayedDisconnectNotification re-verify guard --- + + @Test + public void givenEdgeReconnectedToThisNode_whenDelayFires_thenNotificationSuppressed() { + // A live session exists again on this node - suppress the stale disconnect notification. + sessions().put(edgeId, mock(EdgeGrpcSession.class)); + + fireDelayedDisconnectNotification(); + + verify(ruleProcessor, never()).process(any()); + } + + @Test + public void givenEdgeReconnectedToAnotherNode_whenDelayFires_thenNotificationSuppressed() { + // No local session, but the cluster cache still points at some node: the edge is connected elsewhere. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); + + fireDelayedDisconnectNotification(); + + verify(ruleProcessor, never()).process(any()); + } + + @Test + public void givenEdgeStaysDisconnectedClusterWide_whenDelayFires_thenNotificationSent() { + // No local session and no cache entry on any node: the edge is genuinely down - fire the notification. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); + when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); + + fireDelayedDisconnectNotification(); + + verify(ruleProcessor, times(1)).process(argThat(disconnectTrigger())); + } + + @Test + public void givenPendingDisconnect_whenDestroy_thenNotificationFlushed() { + // The edge is genuinely down (no session, no cache). A graceful shutdown must flush the pending + // notification rather than drop it, otherwise a restart within the delay window swallows the alert. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); + when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); + pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(tenantId, edge, null)); + + destroy(); + + verify(ruleProcessor, times(1)).process(argThat(disconnectTrigger())); + } + + @Test + public void givenPendingDisconnectButReconnectedElsewhere_whenDestroy_thenNotificationSuppressed() { + // The flush still honors the re-verify guard: an edge that reconnected to another node must not alert. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); + pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(tenantId, edge, null)); + + destroy(); + + verify(ruleProcessor, never()).process(any()); + } + + private void destroy() { + ReflectionTestUtils.invokeMethod(edgeGrpcService, "destroy"); + } + + private void evictServiceIdCacheIfOwnedByThisNode() { + ReflectionTestUtils.invokeMethod(edgeGrpcService, "evictServiceIdCacheIfOwnedByThisNode", edgeId); + } + + private void fireDelayedDisconnectNotification() { + EdgeGrpcService.PendingDisconnect pending = new EdgeGrpcService.PendingDisconnect(tenantId, edge, null); + ReflectionTestUtils.invokeMethod(edgeGrpcService, "fireDelayedDisconnectNotification", pending); + } + + @SuppressWarnings("unchecked") + private ConcurrentMap sessions() { + return (ConcurrentMap) ReflectionTestUtils.getField(edgeGrpcService, "sessions"); + } + + @SuppressWarnings("unchecked") + private ConcurrentMap pendingDisconnects() { + return (ConcurrentMap) ReflectionTestUtils.getField(edgeGrpcService, "pendingDisconnectNotifications"); + } + + private static ArgumentMatcher disconnectTrigger() { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger && !edgeTrigger.isConnected(); + } + +} diff --git a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java index 71b4925901..47934b4df0 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java +++ b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java @@ -69,7 +69,15 @@ public class ThingsBoardExecutors { } public static ScheduledExecutorService newSingleThreadScheduledExecutor(String name) { - return Executors.unconfigurableScheduledExecutorService(new ThingsBoardScheduledThreadPoolExecutor(1, ThingsBoardThreadFactory.forName(name))); + return newSingleThreadScheduledExecutor(name, false); + } + + public static ScheduledExecutorService newSingleThreadScheduledExecutor(String name, boolean removeOnCancelPolicy) { + ThingsBoardScheduledThreadPoolExecutor executor = new ThingsBoardScheduledThreadPoolExecutor(1, ThingsBoardThreadFactory.forName(name)); + // Must be set before wrapping: unconfigurableScheduledExecutorService hides the setter. With it enabled, + // cancelled tasks are removed from the delay queue immediately instead of lingering until their fire time. + executor.setRemoveOnCancelPolicy(removeOnCancelPolicy); + return Executors.unconfigurableScheduledExecutorService(executor); } public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, String name) { From 44ea99b1e199af49a9d1947ad2dbe2e32340e93c Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 10 Jun 2026 17:20:19 +0300 Subject: [PATCH 4/9] guard cache evict, drop holder array, derive tenant from edge --- .../service/edge/rpc/EdgeGrpcService.java | 65 +++++++++++++------ .../service/edge/rpc/EdgeGrpcServiceTest.java | 6 +- 2 files changed, 49 insertions(+), 22 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index b875226034..259f63c206 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -405,7 +405,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i cancelPendingDisconnectNotification(edgeId); // Connect notifies immediately; only the disconnect notification is debounced (see scheduleDisconnectNotification). pushStateEventToRuleEngine(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); - notifyEdgeConnectivity(tenantId, edge, true); + notifyEdgeConnectivity(edge, true); cancelScheduleEdgeEventsCheck(edgeId); edgeEventsMigrationProcessed.putIfAbsent(edgeId, Boolean.FALSE); scheduleEdgeEventsCheck(edgeGrpcSession); @@ -562,7 +562,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i long lastDisconnectTs = System.currentTimeMillis(); save(tenantId, edgeId, LAST_DISCONNECT_TIME, lastDisconnectTs); pushStateEventToRuleEngine(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); - scheduleDisconnectNotification(tenantId, edge); + scheduleDisconnectNotification(edge); cancelScheduleEdgeEventsCheck(edgeId); } else { log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId); @@ -578,7 +578,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId); } } - evictServiceIdCacheIfOwnedByThisNode(edgeId); + // Don't evict while a live session for this edge still exists on this node (e.g. a stale session + // disconnecting after a newer one already replaced it) - that newer session legitimately owns the + // cache entry, and wiping it would let another node's pending task fire a false 'disconnected' notification. + if (!sessions.containsKey(edgeId)) { + evictServiceIdCacheIfOwnedByThisNode(edgeId); + } } // Only evict if the cache still points to this node. If the edge already reconnected to a different @@ -702,22 +707,22 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - private void notifyEdgeConnectivity(TenantId tenantId, Edge edge, boolean connected) { + private void notifyEdgeConnectivity(Edge edge, boolean connected) { try { ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() - .tenantId(tenantId) + .tenantId(edge.getTenantId()) .customerId(edge.getCustomerId()) .edgeId(edge.getId()) .edgeName(edge.getName()) .connected(connected).build()); } catch (Exception e) { - log.warn("[{}][{}] Failed to process edge connectivity notification (connected={})", tenantId, edge.getId(), connected, e); + log.warn("[{}][{}] Failed to process edge connectivity notification (connected={})", edge.getTenantId(), edge.getId(), connected, e); } } - private void scheduleDisconnectNotification(TenantId tenantId, Edge edge) { + private void scheduleDisconnectNotification(Edge edge) { if (disconnectNotificationDelayMs <= 0) { - notifyEdgeConnectivity(tenantId, edge, false); + notifyEdgeConnectivity(edge, false); return; } EdgeId edgeId = edge.getId(); @@ -725,28 +730,30 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i if (existing != null) { cancelIfPending(existing.future()); } - // Single-element holder so the scheduled task can reference its own pending entry, which doesn't exist - // until schedule(...) returns. The task needs it for the identity-keyed remove in fireDelayedDisconnectNotification. - PendingDisconnect[] holder = {null}; + // Create the pending entry first so the scheduled task can reference it (for the identity-keyed + // remove in fireDelayedDisconnectNotification), then back-fill its future. Doing this inside compute() + // keeps it under the map bin lock, so the future is set before the entry becomes visible to other threads. + PendingDisconnect pending = new PendingDisconnect(edge); ScheduledFuture future = executorService.schedule( - () -> fireDelayedDisconnectNotification(holder[0]), + () -> fireDelayedDisconnectNotification(pending), disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); - holder[0] = new PendingDisconnect(tenantId, edge, future); - return holder[0]; + pending.setFuture(future); + return pending; }); } private void fireDelayedDisconnectNotification(PendingDisconnect pending) { - EdgeId edgeId = pending.edge().getId(); + Edge edge = pending.edge(); + EdgeId edgeId = edge.getId(); // Identity-keyed remove: don't clobber a newer entry if a second disconnect races with this task firing. pendingDisconnectNotifications.remove(edgeId, pending); // Re-verify the edge is still disconnected. The cache is cluster-wide, so this also covers the case // where the edge dropped on this node and reconnected to a different TB-Core node within the window. if (sessions.containsKey(edgeId) || edgeIdServiceIdCache.get(edgeId) != null) { - log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", pending.tenantId(), edgeId); + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", edge.getTenantId(), edgeId); return; } - notifyEdgeConnectivity(pending.tenantId(), pending.edge(), false); + notifyEdgeConnectivity(edge, false); } private void cancelPendingDisconnectNotification(EdgeId edgeId) { @@ -757,8 +764,28 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } // Carries the context the delayed task needs, so a pending notification can still be fired on shutdown - // (see destroy()), not just cancelled. Package-private for unit testing of fireDelayedDisconnectNotification. - record PendingDisconnect(TenantId tenantId, Edge edge, ScheduledFuture future) {} + // (see destroy()), not just cancelled. The future is back-filled right after scheduling (see + // scheduleDisconnectNotification). Package-private for unit testing of fireDelayedDisconnectNotification. + static final class PendingDisconnect { + private final Edge edge; + private ScheduledFuture future; + + PendingDisconnect(Edge edge) { + this.edge = edge; + } + + Edge edge() { + return edge; + } + + ScheduledFuture future() { + return future; + } + + void setFuture(ScheduledFuture future) { + this.future = future; + } + } private static void cancelIfPending(ScheduledFuture future) { cancelIfPending(future, false); diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java index f538bd1baa..86758cb68f 100644 --- a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java @@ -156,7 +156,7 @@ public class EdgeGrpcServiceTest { // notification rather than drop it, otherwise a restart within the delay window swallows the alert. when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); - pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(tenantId, edge, null)); + pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(edge)); destroy(); @@ -167,7 +167,7 @@ public class EdgeGrpcServiceTest { public void givenPendingDisconnectButReconnectedElsewhere_whenDestroy_thenNotificationSuppressed() { // The flush still honors the re-verify guard: an edge that reconnected to another node must not alert. when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); - pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(tenantId, edge, null)); + pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(edge)); destroy(); @@ -183,7 +183,7 @@ public class EdgeGrpcServiceTest { } private void fireDelayedDisconnectNotification() { - EdgeGrpcService.PendingDisconnect pending = new EdgeGrpcService.PendingDisconnect(tenantId, edge, null); + EdgeGrpcService.PendingDisconnect pending = new EdgeGrpcService.PendingDisconnect(edge); ReflectionTestUtils.invokeMethod(edgeGrpcService, "fireDelayedDisconnectNotification", pending); } From 9fdb32da0568c57e2930edbf98fd1a591a4c50ea Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 10 Jun 2026 17:59:08 +0300 Subject: [PATCH 5/9] refactor: tidy edge disconnect-notification per review --- .../service/edge/rpc/EdgeGrpcService.java | 31 +++++++++---------- .../edge/EdgeConnectionNotificationTest.java | 6 ++-- .../service/edge/rpc/EdgeGrpcServiceTest.java | 12 +------ 3 files changed, 19 insertions(+), 30 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 259f63c206..411ccd5f05 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -238,18 +238,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @PreDestroy public void destroy() { - // Flush already-pending disconnect notifications BEFORE shutting the server down. These are edges that - // disconnected on their own within the delay window; without this a graceful restart would silently - // swallow their "disconnected" alert. Snapshotting first means we only flush those - not edges that this - // very shutdown is about to disconnect (they rebalance to another node and shouldn't alert) - and we run - // while the scheduler and rule engine are still alive. The guard inside fireDelayedDisconnectNotification - // still suppresses any edge that reconnected elsewhere. List pendingToFlush = new ArrayList<>(pendingDisconnectNotifications.values()); pendingDisconnectNotifications.clear(); - pendingToFlush.forEach(pending -> { - cancelIfPending(pending.future()); - fireDelayedDisconnectNotification(pending); - }); + pendingToFlush.forEach(this::fireDelayedDisconnectNotification); if (server != null) { server.shutdownNow(); } @@ -590,12 +581,22 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i // TB-Core node within the keep-alive window, that node has overwritten the entry - evicting it here // would wipe the live owner and make fireDelayedDisconnectNotification raise a false 'disconnected'. private void evictServiceIdCacheIfOwnedByThisNode(EdgeId edgeId) { - TbCacheValueWrapper wrapper = edgeIdServiceIdCache.get(edgeId); - if (wrapper != null && serviceInfoProvider.getServiceId().equals(wrapper.get())) { + if (isOwnedByThisNode(edgeId)) { edgeIdServiceIdCache.evict(edgeId); } } + // The edge's service-id cache entry still points at this node, i.e. this node is the recorded owner. + private boolean isOwnedByThisNode(EdgeId edgeId) { + TbCacheValueWrapper wrapper = edgeIdServiceIdCache.get(edgeId); + return wrapper != null && serviceInfoProvider.getServiceId().equals(wrapper.get()); + } + + // The edge has a live owner somewhere in the cluster (the cache is cluster-wide), regardless of which node. + private boolean isConnectedClusterWide(EdgeId edgeId) { + return edgeIdServiceIdCache.get(edgeId) != null; + } + private void destroySession(EdgeGrpcSession session) { try (session) { if (!session.destroy()) { @@ -749,7 +750,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i pendingDisconnectNotifications.remove(edgeId, pending); // Re-verify the edge is still disconnected. The cache is cluster-wide, so this also covers the case // where the edge dropped on this node and reconnected to a different TB-Core node within the window. - if (sessions.containsKey(edgeId) || edgeIdServiceIdCache.get(edgeId) != null) { + if (sessions.containsKey(edgeId) || isConnectedClusterWide(edgeId)) { log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", edge.getTenantId(), edgeId); return; } @@ -763,9 +764,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - // Carries the context the delayed task needs, so a pending notification can still be fired on shutdown - // (see destroy()), not just cancelled. The future is back-filled right after scheduling (see - // scheduleDisconnectNotification). Package-private for unit testing of fireDelayedDisconnectNotification. static final class PendingDisconnect { private final Edge edge; private ScheduledFuture future; @@ -785,6 +783,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i void setFuture(ScheduledFuture future) { this.future = future; } + } private static void cancelIfPending(ScheduledFuture future) { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java index c46d82437a..66023c7372 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -43,7 +43,7 @@ import static org.mockito.Mockito.verify; @DaoSqlTest public class EdgeConnectionNotificationTest extends AbstractEdgeTest { - private static final long DELAY_MS = 5000L; + private static final long DELAY_MS = 1500L; @MockitoSpyBean private NotificationRuleProcessor notificationRuleProcessor; @@ -115,8 +115,8 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { // The "disconnected" notification must never be sent throughout the full delay window. await().during(DELAY_MS + 500, TimeUnit.MILLISECONDS) - .atMost(DELAY_MS + 2000, TimeUnit.MILLISECONDS) - .untilAsserted(() -> verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false)))); + .atMost(DELAY_MS + 2000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false)))); } private ArgumentMatcher edgeConnectionTrigger(boolean connected) { diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java index 86758cb68f..61bb04a803 100644 --- a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java @@ -45,12 +45,6 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -/** - * Unit coverage for the cluster-aware predicates that guard the delayed disconnect notification: - * {@code evictServiceIdCacheIfOwnedByThisNode} and the re-verify check in - * {@code fireDelayedDisconnectNotification}. These exercise the cross-node ownership logic that the - * single-node integration tests ({@code EdgeConnectionNotificationTest}) cannot reach. - */ @ExtendWith(MockitoExtension.class) public class EdgeGrpcServiceTest { @@ -72,21 +66,17 @@ public class EdgeGrpcServiceTest { @InjectMocks private EdgeGrpcService edgeGrpcService; - private TenantId tenantId; private EdgeId edgeId; private Edge edge; @BeforeEach public void setUp() { - tenantId = new TenantId(UUID.randomUUID()); edgeId = new EdgeId(UUID.randomUUID()); edge = new Edge(edgeId); - edge.setTenantId(tenantId); + edge.setTenantId(TenantId.fromUUID(UUID.randomUUID())); edge.setName("test-edge"); } - // --- evictServiceIdCacheIfOwnedByThisNode --- - @Test public void givenCacheOwnedByThisNode_whenEvict_thenEntryIsEvicted() { when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); From 668ca3b632674bc42d1616d4de58d18c5e94be26 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 11 Jun 2026 10:55:51 +0300 Subject: [PATCH 6/9] Fixes after dev-rereview --- .../service/edge/rpc/EdgeGrpcService.java | 24 +++++++++++++++---- .../edge/EdgeConnectionNotificationTest.java | 9 +++++-- 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 411ccd5f05..297a45ec2f 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -84,6 +84,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; @@ -722,6 +723,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void scheduleDisconnectNotification(Edge edge) { + // Zero delay means "no debounce": notify immediately and skip the cluster-wide re-verify guard. if (disconnectNotificationDelayMs <= 0) { notifyEdgeConnectivity(edge, false); return; @@ -729,7 +731,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i EdgeId edgeId = edge.getId(); pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { if (existing != null) { - cancelIfPending(existing.future()); + cancelIfPending(existing.getFuture()); } // Create the pending entry first so the scheduled task can reference it (for the identity-keyed // remove in fireDelayedDisconnectNotification), then back-fill its future. Doing this inside compute() @@ -744,7 +746,13 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void fireDelayedDisconnectNotification(PendingDisconnect pending) { - Edge edge = pending.edge(); + // Claim-once guard: the @PreDestroy destroy() flush and a concurrently-firing scheduled task can both call + // this for the same PendingDisconnect. tryClaim() ensures notifyEdgeConnectivity fires at most once without + // relying on downstream notification dedup. + if (!pending.tryClaim()) { + return; + } + Edge edge = pending.getEdge(); EdgeId edgeId = edge.getId(); // Identity-keyed remove: don't clobber a newer entry if a second disconnect races with this task firing. pendingDisconnectNotifications.remove(edgeId, pending); @@ -760,23 +768,25 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void cancelPendingDisconnectNotification(EdgeId edgeId) { PendingDisconnect pending = pendingDisconnectNotifications.remove(edgeId); if (pending != null) { - cancelIfPending(pending.future()); + cancelIfPending(pending.getFuture()); } } static final class PendingDisconnect { + private final Edge edge; + private final AtomicBoolean notified = new AtomicBoolean(false); private ScheduledFuture future; PendingDisconnect(Edge edge) { this.edge = edge; } - Edge edge() { + Edge getEdge() { return edge; } - ScheduledFuture future() { + ScheduledFuture getFuture() { return future; } @@ -784,6 +794,10 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i this.future = future; } + boolean tryClaim() { + return notified.compareAndSet(false, true); + } + } private static void cancelIfPending(ScheduledFuture future) { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java index 66023c7372..4482177750 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -30,6 +30,7 @@ import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.edge.imitator.EdgeImitator; import org.thingsboard.server.service.edge.rpc.EdgeGrpcService; +import java.util.Map; import java.util.concurrent.TimeUnit; import static org.awaitility.Awaitility.await; @@ -101,8 +102,12 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { // Edge drops... edgeImitator.disconnect(); - // Ensure the server processed the disconnect (and scheduled the delayed notification) before reconnecting. - TimeUnit.SECONDS.sleep(1); + // Wait until the server has processed the disconnect and scheduled the pending notification + // (the edge id appears in the pendingDisconnectNotifications map) before reconnecting. + await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).until(() -> { + Map pending = (Map) ReflectionTestUtils.getField(edgeGrpcService, "pendingDisconnectNotifications"); + return pending != null && pending.containsKey(edge.getId()); + }); // ...and reconnects within the delay window, which must cancel the pending "disconnected" notification. EdgeImitator reconnected = createEdgeImitator(); From bad121985105903340506ab693cc02231cfd3bc4 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 19 Aug 2026 12:13:46 +0300 Subject: [PATCH 7/9] Compare edge customer before session state is replaced in onConfigurationUpdate --- .../thingsboard/server/service/edge/rpc/EdgeGrpcSession.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 43a4417323..43dc153c91 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventType; +import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; @@ -243,8 +244,9 @@ public abstract class EdgeGrpcSession implements Closeable { public void onConfigurationUpdate(Edge edge) { log.debug("[{}] onConfigurationUpdate [{}]", sessionId, edge); this.tenantId = edge.getTenantId(); + CustomerId stateCustomerId = this.edge != null ? this.edge.getCustomerId() : null; this.edge = edge; - if (!this.edge.getCustomerId().equals(edge.getCustomerId())) { + if (stateCustomerId != null && !stateCustomerId.equals(edge.getCustomerId())) { // do not send edge configuration message on customer update // message send by separate flow from assign_to or unassing_from customer return; From 1ad0c481dd74fb8e8fca212d4377abe8159fb26a Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 19 Aug 2026 13:05:35 +0300 Subject: [PATCH 8/9] Do not strand edge sync when an event fetcher fails --- .../server/service/edge/rpc/EdgeGrpcSession.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 1e9a0ade87..aeab041c7f 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -269,7 +269,12 @@ public abstract class EdgeGrpcSession implements Closeable { @Override public void onFailure(Throwable t) { - log.error("[{}][{}] Exception during sync process", tenantId, edge.getId(), t); + log.error("[{}][{}] Exception during sync process, skipping fetcher {} and continuing", + tenantId, edge.getId(), next.getClass().getSimpleName(), t); + // Keep walking the cursor: returning here leaves syncInProgress set for the life of the + // session, so the edge never receives SyncCompletedMsg and both general downlink delivery + // and uplink processing stay gated until the session is re-established. + doSync(cursor); } }, ctx.getGrpcCallbackExecutorService()); } else { From 966aea28bb90291438a3fb8567296f26fde553d4 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 19 Aug 2026 13:05:35 +0300 Subject: [PATCH 9/9] Do not strand edge sync when an event fetcher fails --- .../server/service/edge/rpc/EdgeGrpcSession.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 43a4417323..d1b21f7110 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -282,7 +282,12 @@ public abstract class EdgeGrpcSession implements Closeable { @Override public void onFailure(Throwable t) { - log.error("[{}][{}] Exception during sync process", tenantId, edge.getId(), t); + log.error("[{}][{}] Exception during sync process, skipping fetcher {} and continuing", + tenantId, edge.getId(), next.getClass().getSimpleName(), t); + // Keep walking the cursor: returning here leaves syncInProgress set for the life of the + // session, so the edge never receives SyncCompletedMsg and both general downlink delivery + // and uplink processing stay gated until the session is re-established. + doSync(cursor); } }, ctx.getGrpcCallbackExecutorService()); } else {