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..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 @@ -36,12 +36,12 @@ 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; 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 +58,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; @@ -83,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; @@ -103,6 +105,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,10 +139,16 @@ 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; + @Autowired + private TelemetrySubscriptionService tsSubService; + @Autowired private TbClusterService clusterService; @@ -194,7 +203,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!"); } @@ -228,17 +239,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @PreDestroy public void destroy() { + List pendingToFlush = new ArrayList<>(pendingDisconnectNotifications.values()); + pendingDisconnectNotifications.clear(); + pendingToFlush.forEach(this::fireDelayedDisconnectNotification); 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); - } - } + sessionEdgeEventChecks.values().forEach(task -> cancelIfPending(task, true)); + sessionEdgeEventChecks.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); } @@ -323,6 +331,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } cancelScheduleEdgeEventsCheck(edgeId); } + cancelPendingDisconnectNotification(edgeId); } private void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) { @@ -383,7 +392,12 @@ 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); + // Connect notifies immediately; only the disconnect notification is debounced (see scheduleDisconnectNotification). + pushStateEventToRuleEngine(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); + notifyEdgeConnectivity(edge, true); cancelScheduleEdgeEventsCheck(edgeId); edgeEventsMigrationProcessed.putIfAbsent(edgeId, Boolean.FALSE); scheduleEdgeEventsCheck(edgeGrpcSession); @@ -517,13 +531,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) { @@ -545,7 +553,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(edge); cancelScheduleEdgeEventsCheck(edgeId); } else { log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId); @@ -561,7 +570,32 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId); } } - edgeIdServiceIdCache.evict(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 + // 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) { + 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) { @@ -580,14 +614,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) @@ -600,14 +634,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) @@ -617,7 +651,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 +689,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 +709,107 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } + private void notifyEdgeConnectivity(Edge edge, boolean connected) { + try { + ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() + .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={})", edge.getTenantId(), edge.getId(), connected, e); + } + } + + 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; + } + EdgeId edgeId = edge.getId(); + pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { + if (existing != null) { + 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() + // 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(pending), + disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); + pending.setFuture(future); + return pending; + }); + } + + private void fireDelayedDisconnectNotification(PendingDisconnect pending) { + // 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); + // 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) || isConnectedClusterWide(edgeId)) { + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", edge.getTenantId(), edgeId); + return; + } + notifyEdgeConnectivity(edge, false); + } + + private void cancelPendingDisconnectNotification(EdgeId edgeId) { + PendingDisconnect pending = pendingDisconnectNotifications.remove(edgeId); + if (pending != null) { + 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 getEdge() { + return edge; + } + + ScheduledFuture getFuture() { + return future; + } + + void setFuture(ScheduledFuture future) { + this.future = future; + } + + boolean tryClaim() { + return notified.compareAndSet(false, true); + } + + } + + private static void cancelIfPending(ScheduledFuture future) { + cancelIfPending(future, false); + } + + private static void cancelIfPending(ScheduledFuture future, boolean mayInterruptIfRunning) { + if (future != null && !future.isDone()) { + future.cancel(mayInterruptIfRunning); + } + } + private void cleanupZombieSessions() { try { tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId())); @@ -707,6 +862,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i !kafkaSession.getConsumer().getConsumer().isStopped(); } return false; + } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 9b0ca8d847..7a2e002750 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1672,6 +1672,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: "${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..8a3c3c1ae7 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -157,16 +157,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 new file mode 100644 index 0000000000..4482177750 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -0,0 +1,133 @@ +/** + * 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.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentMatcher; +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.Map; +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 +public class EdgeConnectionNotificationTest extends AbstractEdgeTest { + + private static final long DELAY_MS = 1500L; + + @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(); + + 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... + edgeImitator.disconnect(); + // 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(); + 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)))); + + // 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) { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger + && edge.getId().equals(edgeTrigger.getEdgeId()) + && edgeTrigger.isConnected() == 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 new file mode 100644 index 0000000000..61bb04a803 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java @@ -0,0 +1,194 @@ +/** + * 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; + +@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 EdgeId edgeId; + private Edge edge; + + @BeforeEach + public void setUp() { + edgeId = new EdgeId(UUID.randomUUID()); + edge = new Edge(edgeId); + edge.setTenantId(TenantId.fromUUID(UUID.randomUUID())); + edge.setName("test-edge"); + } + + @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(edge)); + + 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(edge)); + + 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(edge); + 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) {