From 00238968e9dd62a5a3504f75c97ba52e774d0fc0 Mon Sep 17 00:00:00 2001
From: Dmytro Skarzhynets <dskarzhynets@thingsboard.io>
Date: Tue, 24 Feb 2026 20:34:31 +0200
Subject: [PATCH] fix: preserve rule node execution counter when copying
 messages in delay and deduplication nodes

Delay and deduplication rule nodes were creating brand new TbMsg objects
instead of copying the original, which reset the ruleNodeExecCounter to 0.
This allowed bypassing the maxRuleNodeExecutionsPerMessage limit.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 .../RuleNodeActorMessageProcessor.java        |   4 +-
 .../thingsboard/server/common/msg/TbMsg.java  |   5 +-
 .../deduplication/TbMsgDeduplicationNode.java |  15 +--
 .../rule/engine/delay/TbMsgDelayNode.java     |  14 +--
 .../external/TbAbstractExternalNode.java      |   8 +-
 .../rule/engine/delay/TbMsgDelayNodeTest.java | 106 ++++++++++++++++++
 .../transform/TbMsgDeduplicationNodeTest.java |  13 +++
 7 files changed, 136 insertions(+), 29 deletions(-)
 create mode 100644 rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java

diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
index 3eecae7f7a..cc74d248d2 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
@@ -137,7 +137,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
                 defaultCtx.tellFailure(msg.getMsg(), e);
             }
         } else {
-            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more than " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
         }
     }
 
@@ -160,7 +160,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
                     msg.getCtx().tellFailure(msg.getMsg(), e);
                 }
             } else {
-                tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+                tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more than " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
             }
         }
     }
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
index 831d0557cc..c2cd6eb9b9 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
@@ -110,11 +110,10 @@ public final class TbMsg implements Serializable {
                 .build();
     }
 
-    public TbMsg copyWithNewCtx() {
+    public TbMsgBuilder copyWithNewCtx() {
         return copy()
                 .ctx(ctx.copy())
-                .callback(TbMsgCallback.EMPTY)
-                .build();
+                .callback(TbMsgCallback.EMPTY);
     }
 
     private TbMsg(String queueName, UUID id, long ts, TbMsgType internalType, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java
index b4d2951264..9a29f80b1c 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java
@@ -41,6 +41,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
@@ -177,15 +178,11 @@ public class TbMsgDeduplicationNode implements TbNode {
                         }
                     }
                     if (resultMsg != null) {
-                        String queueName1 = queueName != null ? queueName : resultMsg.getQueueName();
-                        deduplicationResults.add(TbMsg.newMsg()
-                                .queueName(queueName1)
-                                .type(resultMsg.getType())
-                                .originator(resultMsg.getOriginator())
-                                .customerId(resultMsg.getCustomerId())
-                                .copyMetaData(resultMsg.getMetaData())
-                                .data(resultMsg.getData())
-                                .build());
+                        var msgBuilder = resultMsg.copyWithNewCtx().id(UUID.randomUUID());
+                        if (queueName != null) {
+                            msgBuilder.queueName(queueName);
+                        }
+                        deduplicationResults.add(msgBuilder.build());
                     }
                 }
                 packBoundsOpt = findValidPack(msgList, deduplicationTimeoutMs);
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
index f0e4a3e40e..04288cf7f3 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
@@ -62,17 +62,9 @@ public class TbMsgDelayNode implements TbNode {
         if (msg.isTypeOf(TbMsgType.DELAY_TIMEOUT_SELF_MSG)) {
             TbMsg pendingMsg = pendingMsgs.remove(UUID.fromString(msg.getData()));
             if (pendingMsg != null) {
-                ctx.enqueueForTellNext(
-                        TbMsg.newMsg()
-                                .queueName(pendingMsg.getQueueName())
-                                .type(pendingMsg.getType())
-                                .originator(pendingMsg.getOriginator())
-                                .customerId(pendingMsg.getCustomerId())
-                                .copyMetaData(pendingMsg.getMetaData())
-                                .data(pendingMsg.getData())
-                                .build(),
-                        TbNodeConnectionType.SUCCESS
-                );
+                ctx.enqueueForTellNext(pendingMsg.copyWithNewCtx()
+                        .id(UUID.randomUUID())
+                        .build(), TbNodeConnectionType.SUCCESS);
             }
         } else {
             if (pendingMsgs.size() < config.getMaxPendingMsgs()) {
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java
index e2c218f272..4efa462e37 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java
@@ -30,7 +30,7 @@ public abstract class TbAbstractExternalNode implements TbNode {
 
     protected void tellSuccess(TbContext ctx, TbMsg tbMsg) {
         if (forceAck) {
-            ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.SUCCESS);
+            ctx.enqueueForTellNext(tbMsg.copyWithNewCtx().build(), TbNodeConnectionType.SUCCESS);
         } else {
             ctx.tellSuccess(tbMsg);
         }
@@ -39,9 +39,9 @@ public abstract class TbAbstractExternalNode implements TbNode {
     protected void tellFailure(TbContext ctx, TbMsg tbMsg, Throwable t) {
         if (forceAck) {
             if (t == null) {
-                ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.FAILURE);
+                ctx.enqueueForTellNext(tbMsg.copyWithNewCtx().build(), TbNodeConnectionType.FAILURE);
             } else {
-                ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
+                ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx().build(), t);
             }
         } else {
             if (t == null) {
@@ -55,7 +55,7 @@ public abstract class TbAbstractExternalNode implements TbNode {
     protected TbMsg ackIfNeeded(TbContext ctx, TbMsg msg) {
         if (forceAck) {
             ctx.ack(msg);
-            return msg.copyWithNewCtx();
+            return msg.copyWithNewCtx().build();
         } else {
             return msg;
         }
diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java
new file mode 100644
index 0000000000..7e3220cdfb
--- /dev/null
+++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java
@@ -0,0 +1,106 @@
+/**
+ * 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.rule.engine.delay;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.TbMsgProcessingCtx;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.lenient;
+
+@ExtendWith(MockitoExtension.class)
+class TbMsgDelayNodeTest {
+
+    final DeviceId deviceId = new DeviceId(UUID.fromString("5770153d-6ca2-4447-8a54-5d8a4538e052"));
+    final RuleNodeId ruleNodeId = new RuleNodeId(UUID.fromString("ee682a85-7f5a-4182-91bc-46e555138fe2"));
+
+    TbMsgDelayNode node;
+
+    @Mock
+    TbContext ctxMock;
+
+    @BeforeEach
+    void setUp() throws TbNodeException {
+        node = new TbMsgDelayNode();
+        var config = new TbMsgDelayNodeConfiguration().defaultConfiguration();
+        node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
+
+        lenient().when(ctxMock.getSelfId()).thenReturn(ruleNodeId);
+    }
+
+    @Test
+    void shouldPreserveRuleNodeCounterAndResetCallbackWhenEnqueuingDelayedMsg() {
+        // GIVEN
+        int ruleNodeExecCounter = 5;
+        var originalMsg = TbMsg.newMsg()
+                .id(UUID.randomUUID())
+                .type(TbMsgType.POST_TELEMETRY_REQUEST)
+                .originator(deviceId)
+                .metaData(TbMsgMetaData.EMPTY)
+                .data("{\"temperature\":42}")
+                .ctx(new TbMsgProcessingCtx(ruleNodeExecCounter))
+                .build();
+
+        String originalMsgId = originalMsg.getId().toString();
+        var tickMsg = TbMsg.newMsg()
+                .type(TbMsgType.DELAY_TIMEOUT_SELF_MSG)
+                .originator(ruleNodeId)
+                .metaData(TbMsgMetaData.EMPTY)
+                .data(originalMsgId)
+                .build();
+        given(ctxMock.newMsg(null, TbMsgType.DELAY_TIMEOUT_SELF_MSG, ruleNodeId, null, TbMsgMetaData.EMPTY, originalMsgId)).willReturn(tickMsg);
+
+        node.onMsg(ctxMock, originalMsg);
+
+        // WHEN
+        node.onMsg(ctxMock, tickMsg);
+
+        // THEN
+        var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        then(ctxMock).should().enqueueForTellNext(msgCaptor.capture(), eq(TbNodeConnectionType.SUCCESS));
+
+        var enqueuedMsg = msgCaptor.getValue();
+        assertThat(enqueuedMsg).usingRecursiveComparison()
+                .ignoringFields("id", "ts", "callback")
+                .isEqualTo(originalMsg);
+
+        assertThat(enqueuedMsg.getId()).isNotNull().isNotEqualTo(originalMsg.getId());
+        assertThat(enqueuedMsg.getAndIncrementRuleNodeCounter()).isEqualTo(ruleNodeExecCounter);
+        assertThat(enqueuedMsg.getCallback()).isSameAs(TbMsgCallback.EMPTY);
+    }
+
+}
diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java
index 1baae125ef..ec71819878 100644
--- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java
+++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java
@@ -47,6 +47,8 @@ import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.TbMsgProcessingCtx;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -79,6 +81,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
 
     private TbContext ctx;
 
+    private static final int RULE_NODE_EXEC_COUNTER = 5;
+
     private final ScheduledExecutorService executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("de-duplication-node-test");
     private final int deduplicationInterval = 1;
 
@@ -204,6 +208,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(firstMsg.getData(), actualMsg.getData());
         Assertions.assertEquals(firstMsg.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(firstMsg.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
 
         if (queueName == null) {
             Assertions.assertEquals(firstMsg.getQueueName(), actualMsg.getQueueName());
@@ -257,6 +263,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(msgWithLatestTs.getData(), actualMsg.getData());
         Assertions.assertEquals(msgWithLatestTs.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(msgWithLatestTs.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
     }
 
     @Test
@@ -402,6 +410,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(msgWithLatestTsInFirstPack.getData(), actualMsg.getData());
         Assertions.assertEquals(msgWithLatestTsInFirstPack.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(msgWithLatestTsInFirstPack.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
 
         // verify that newMsg is called but content of messages is the same as in the last msg for the second pack.
         actualMsg = resultMsgs.get(1);
@@ -411,6 +421,8 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
         Assertions.assertEquals(msgWithLatestTsInSecondPack.getData(), actualMsg.getData());
         Assertions.assertEquals(msgWithLatestTsInSecondPack.getMetaData(), actualMsg.getMetaData());
         Assertions.assertEquals(msgWithLatestTsInSecondPack.getType(), actualMsg.getType());
+        Assertions.assertEquals(RULE_NODE_EXEC_COUNTER, actualMsg.getAndIncrementRuleNodeCounter());
+        Assertions.assertSame(TbMsgCallback.EMPTY, actualMsg.getCallback());
     }
 
     @Test
@@ -539,6 +551,7 @@ public class TbMsgDeduplicationNodeTest extends AbstractRuleNodeUpgradeTest {
                 .originator(deviceId)
                 .copyMetaData(metaData)
                 .data(JacksonUtil.toString(dataNode))
+                .ctx(new TbMsgProcessingCtx(RULE_NODE_EXEC_COUNTER))
                 .build();
     }