diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index bde1873627..e3a69be624 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1676,6 +1676,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/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) {